diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index d8b87bfe7b8..f9ed239db82 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -153,12 +153,15 @@ jobs: run: echo "VERSION=${GITHUB_REF/refs\/tags\//}" >> "$GITHUB_OUTPUT" - name: Create a tag on ZenML Cloud plugins repo uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0 + env: + VERSION: ${{ steps.get_version.outputs.VERSION }} + SHA: ${{ steps.get_sha.outputs.sha }} with: github-token: ${{ secrets.CLOUD_PLUGINS_REPO_PAT }} script: |- await github.rest.git.createRef({ owner: 'zenml-io', repo: 'zenml-cloud-plugins', - ref: 'refs/tags/${{ steps.get_version.outputs.VERSION }}', - sha: '${{ steps.get_sha.outputs.sha }}' + ref: `refs/tags/${process.env.VERSION}`, + sha: process.env.SHA }) diff --git a/.github/zizmor.yml b/.github/zizmor.yml index c0e3302da88..9878e3a265c 100644 --- a/.github/zizmor.yml +++ b/.github/zizmor.yml @@ -12,10 +12,10 @@ rules: # These are ZenML's own templates and are intentionally unpinned unpinned-uses: ignore: - - update-templates-to-examples.yml:44 # zenml-io/template-e2e-batch + - update-templates-to-examples.yml:46 # zenml-io/template-e2e-batch - update-templates-to-examples.yml:120 # zenml-io/template-nlp - - update-templates-to-examples.yml:195 # zenml-io/template-starter - - update-templates-to-examples.yml:272 # zenml-io/template-llm-finetuning + - update-templates-to-examples.yml:193 # zenml-io/template-starter + - update-templates-to-examples.yml:268 # zenml-io/template-llm-finetuning - weekly-agent-pipelines-test.yml:101 # Ilshidur/action-discord uses @master, can't be SHA-pinned # excessive-permissions: These workflows use default permissions. @@ -48,12 +48,9 @@ rules: disable: true # Detect dangerous template expansions - # Ignores: Low-confidence findings for step outputs used in github-script - # (step outputs within same workflow are trusted) + # Ignores: Low-confidence findings for trusted values used in github-script template-injection: ignore: - - release.yml:169 # VERSION from step output - - release.yml:170 # sha from step output - snack-it.yml:352 # issue_number from step output - snack-it.yml:353 # continuation of above - snack-it.yml:354 # project_added from step output diff --git a/src/zenml/cli/__init__.py b/src/zenml/cli/__init__.py index acf4e16dc69..e1a0fdab9f2 100644 --- a/src/zenml/cli/__init__.py +++ b/src/zenml/cli/__init__.py @@ -2591,3 +2591,4 @@ def my_pipeline(...): from zenml.cli.project import * # noqa from zenml.cli.tag import * # noqa from zenml.cli.trigger import * # noqa +from zenml.cli.webhook_integration import * # noqa diff --git a/src/zenml/cli/webhook_integration.py b/src/zenml/cli/webhook_integration.py new file mode 100644 index 00000000000..5c54f11b229 --- /dev/null +++ b/src/zenml/cli/webhook_integration.py @@ -0,0 +1,185 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""CLI commands for webhooks.""" + +from typing import Any + +import click + +from zenml.cli import utils as cli_utils +from zenml.cli.cli import TagGroup, cli +from zenml.cli.utils import OutputFormat, list_options +from zenml.client import Client +from zenml.console import console +from zenml.enums import CliCategories, WebhookType +from zenml.models import WebhookIntegrationFilter + + +@cli.group("webhook", cls=TagGroup, tag=CliCategories.MANAGEMENT_TOOLS) +def webhook() -> None: + """Manage external webhook intake endpoints.""" + + +@webhook.command("create") +@click.argument("name", type=str) +@click.option( + "--type", + "webhook_type", + type=click.Choice([value.value for value in WebhookType]), + required=True, +) +@click.option( + "--secret", + type=str, + default=None, + help="Set a signing secret.", +) +@click.option("--inactive", is_flag=True, default=False) +def create_webhook( + name: str, + webhook_type: str, + secret: str | None, + inactive: bool, +) -> None: + """Create a webhook. + + Args: + name: The webhook name. + webhook_type: The webhook provider type. + secret: An optional user-supplied signing secret. + inactive: Whether to create the webhook inactive. + """ + result = Client().create_webhook( + name=name, + webhook_type=WebhookType(webhook_type), + active=not inactive, + secret=secret, + ) + cli_utils.print_pydantic_model( + title=f"Webhook '{name}'", + model=result.webhook, + ) + if result.secret is not None: + cli_utils.declare( + f"Signing secret: {result.secret.get_secret_value()}" + ) + + +@webhook.command("describe") +@click.argument("name_or_id", type=str) +def describe_webhook(name_or_id: str) -> None: + """Describe a webhook. + + Args: + name_or_id: The webhook name or ID. + """ + integration = Client().get_webhook(name_or_id) + cli_utils.print_pydantic_model( + title=f"Webhook '{integration.name}'", + model=integration, + ) + cli_utils.declare(f"Endpoint path: {integration.endpoint_path}") + + +@webhook.command("list") +@list_options( + WebhookIntegrationFilter, + default_columns=["id", "name", "webhook_type", "active"], +) +def list_webhooks( + columns: str, output_format: OutputFormat, **kwargs: Any +) -> None: + """List webhooks. + + Args: + columns: The columns to display. + output_format: The output format. + **kwargs: The webhook filters. + """ + with console.status("Listing webhooks...\n"): + integrations = Client().list_webhooks(**kwargs) + cli_utils.print_page( + integrations, + columns, + output_format, + empty_message="No webhooks found for this filter.", + ) + + +@webhook.command("update") +@click.argument("name_or_id", type=str) +@click.option("--name", "new_name", type=str, default=None) +@click.option( + "--active/--inactive", "active", default=None, help="Set active state." +) +def update_webhook( + name_or_id: str, + new_name: str | None, + active: bool | None, +) -> None: + """Update a webhook. + + Args: + name_or_id: The webhook name or ID. + new_name: The new webhook name. + active: The new active state. + """ + integration = Client().update_webhook( + name_id_or_prefix=name_or_id, + name=new_name, + active=active, + ) + cli_utils.print_pydantic_model( + title=f"Webhook '{integration.name}'", + model=integration, + ) + + +@webhook.command("rotate-secret") +@click.argument("name_or_id", type=str) +@click.option( + "--secret", + type=str, + default=None, + help="Set an optional direct replacement secret.", +) +def rotate_webhook_secret(name_or_id: str, secret: str | None) -> None: + """Rotate a webhook signing secret. + + Args: + name_or_id: The webhook name or ID. + secret: An optional direct replacement secret. + """ + result = Client().rotate_webhook_secret( + name_id_or_prefix=name_or_id, secret=secret + ) + cli_utils.declare(f"Signing secret: {result.secret.get_secret_value()}") + + +@webhook.command("delete") +@click.argument("name_or_id", type=str) +@click.option("--yes", "confirmed", is_flag=True) +def delete_webhook(name_or_id: str, confirmed: bool) -> None: + """Delete a webhook. + + Args: + name_or_id: The webhook name or ID. + confirmed: Whether to skip the confirmation prompt. + """ + if not confirmed and not cli_utils.confirmation( + f"Delete webhook `{name_or_id}`?" + ): + return + Client().delete_webhook(name_or_id) + cli_utils.declare(f"Deleted webhook `{name_or_id}`.") diff --git a/src/zenml/client.py b/src/zenml/client.py index 5505b7a71cf..56ebfa42e09 100644 --- a/src/zenml/client.py +++ b/src/zenml/client.py @@ -78,6 +78,7 @@ TriggerRunConcurrency, TriggerType, VisualizationResourceTypes, + WebhookType, ) from zenml.exceptions import ( AuthorizationException, @@ -215,6 +216,13 @@ UserRequest, UserResponse, UserUpdate, + WebhookIntegrationCreateResponse, + WebhookIntegrationFilter, + WebhookIntegrationRequest, + WebhookIntegrationResponse, + WebhookIntegrationRotateSecretRequest, + WebhookIntegrationSecretResponse, + WebhookIntegrationUpdate, ) from zenml.models.v2.base.filter import ( DatetimeFilterOption, @@ -6832,6 +6840,202 @@ def delete_code_repository( ) self.zen_store.delete_code_repository(code_repository_id=repo.id) + # ------------------------------ Webhooks ------------------------------ + + @_fail_for_sql_zen_store + def create_webhook( + self, + name: str, + webhook_type: WebhookType, + active: bool = True, + secret: str | None = None, + ) -> WebhookIntegrationCreateResponse: + """Create a webhook in the active project. + + Args: + name: The webhook name. + webhook_type: The webhook provider type. + active: Whether the integration accepts events immediately. + secret: An optional signing secret. + + Returns: + The created webhook and any generated signing secret. + """ + return self.zen_store.create_webhook_integration( + WebhookIntegrationRequest( + project=self.active_project.id, + name=name, + webhook_type=webhook_type, + active=active, + secret=secret, + ) + ) + + @_fail_for_sql_zen_store + def get_webhook( + self, + name_id_or_prefix: str | UUID, + allow_name_prefix_match: bool = True, + project: str | UUID | None = None, + hydrate: bool = True, + ) -> WebhookIntegrationResponse: + """Get a webhook by name, ID, or prefix. + + Args: + name_id_or_prefix: The webhook name, ID, or ID prefix. + allow_name_prefix_match: Whether to allow name prefix matching. + project: The project name or ID. + hydrate: Whether to include intake statistics. + + Returns: + The webhook. + """ + return self._get_entity_by_id_or_name_or_prefix( + get_method=self.zen_store.get_webhook_integration, + list_method=self.list_webhooks, + name_id_or_prefix=name_id_or_prefix, + allow_name_prefix_match=allow_name_prefix_match, + hydrate=hydrate, + project=project, + ) + + @_fail_for_sql_zen_store + def list_webhooks( + self, + sort_by: str = "created", + page: int = PAGINATION_STARTING_PAGE, + size: int = PAGE_SIZE_DEFAULT, + logical_operator: LogicalOperators = LogicalOperators.AND, + id: UUIDFilterOption = None, + created: DatetimeFilterOption = None, + updated: DatetimeFilterOption = None, + name: StringFilterOption = None, + project: str | UUID | None = None, + user: UUIDFilterOption = None, + webhook_type: EnumFilterOption[WebhookType] = None, + active: bool | None = None, + hydrate: bool = False, + ) -> Page[WebhookIntegrationResponse]: + """List webhooks. + + Args: + sort_by: The field used to sort results. + page: The page index. + size: The maximum page size. + logical_operator: The operator used to combine filters. + id: Filter by webhook ID. + created: Filter by creation time. + updated: Filter by update time. + name: Filter by webhook name. + project: Filter by project name or ID. + user: Filter by creating user. + webhook_type: Filter by webhook provider type. + active: Filter by active state. + hydrate: Whether to include intake statistics. + + Returns: + A page of webhooks. + """ + filter_model = WebhookIntegrationFilter( + sort_by=sort_by, + page=page, + size=size, + logical_operator=logical_operator, + id=id, + created=created, + updated=updated, + name=name, + project=project or self.active_project.id, + user=user, + webhook_type=webhook_type, + active=active, + ) + return self.zen_store.list_webhook_integrations( + filter_model=filter_model, hydrate=hydrate + ) + + @_fail_for_sql_zen_store + def update_webhook( + self, + name_id_or_prefix: str | UUID, + name: str | None = None, + active: bool | None = None, + project: str | UUID | None = None, + ) -> WebhookIntegrationResponse: + """Update a webhook. + + Args: + name_id_or_prefix: The webhook name, ID, or ID prefix. + name: The new webhook name. + active: The new active state. + project: The project name or ID. + + Returns: + The updated webhook. + """ + if is_valid_uuid(name_id_or_prefix): + integration_id = ( + UUID(name_id_or_prefix) + if isinstance(name_id_or_prefix, str) + else name_id_or_prefix + ) + else: + integration_id = self.get_webhook( + name_id_or_prefix=name_id_or_prefix, + allow_name_prefix_match=False, + project=project, + ).id + return self.zen_store.update_webhook_integration( + integration_id=integration_id, + update=WebhookIntegrationUpdate(name=name, active=active), + ) + + @_fail_for_sql_zen_store + def delete_webhook( + self, + name_id_or_prefix: str | UUID, + project: str | UUID | None = None, + ) -> None: + """Delete a webhook. + + Args: + name_id_or_prefix: The webhook name, ID, or ID prefix. + project: The project name or ID. + """ + integration = self.get_webhook( + name_id_or_prefix=name_id_or_prefix, + allow_name_prefix_match=False, + project=project, + ) + self.zen_store.delete_webhook_integration(integration.id) + + @_fail_for_sql_zen_store + def rotate_webhook_secret( + self, + name_id_or_prefix: str | UUID, + secret: str | None = None, + project: str | UUID | None = None, + ) -> WebhookIntegrationSecretResponse: + """Rotate a webhook signing secret. + + Args: + name_id_or_prefix: The webhook name, ID, or ID prefix. + secret: An optional direct replacement secret. + project: The project name or ID. + + Returns: + The newly active signing secret. + """ + integration = self.get_webhook( + name_id_or_prefix=name_id_or_prefix, + allow_name_prefix_match=False, + project=project, + ) + return self.zen_store.rotate_webhook_integration_secret( + integration_id=integration.id, + request=WebhookIntegrationRotateSecretRequest(secret=secret), + ) + # --------------------------- Service Connectors --------------------------- def create_service_connector( diff --git a/src/zenml/constants.py b/src/zenml/constants.py index f4b15be4b8c..83b8cccd037 100644 --- a/src/zenml/constants.py +++ b/src/zenml/constants.py @@ -544,6 +544,7 @@ def handle_float_env_var(var: str, default: float = 0.0) -> float: TAGS = "/tags" TAG_RESOURCES = "/tag_resources" TRIGGERS = "/triggers" +WEBHOOKS = "/webhooks" TRIGGER_SNAPSHOT_DISPATCH_STATE = "/dispatch_state" ONBOARDING_STATE = "/onboarding_state" USERS = "/users" diff --git a/src/zenml/enums.py b/src/zenml/enums.py index 831de6c97fc..9dfbcb7c05c 100644 --- a/src/zenml/enums.py +++ b/src/zenml/enums.py @@ -692,6 +692,13 @@ class TriggerRunConcurrency(StrEnum): SUBMIT = "submit" +class WebhookType(StrEnum): + """Supported webhook integration provider types.""" + + GITHUB = "github" + CUSTOM = "custom" + + class ContainerEngineType(StrEnum): """Container engine types.""" diff --git a/src/zenml/models/__init__.py b/src/zenml/models/__init__.py index c354691aca7..b8aab97ab41 100644 --- a/src/zenml/models/__init__.py +++ b/src/zenml/models/__init__.py @@ -381,9 +381,9 @@ StepRunUpdate, ) from zenml.models.v2.core.stream_event import ( - StreamEvent, StreamBatchRequest, StreamBatchResponse, + StreamEvent, ) from zenml.models.v2.core.tag import ( TagFilter, @@ -432,6 +432,20 @@ UserResponseMetadata, UserUpdate, ) +from zenml.models.v2.core.webhook_integration import ( + WebhookEventStatsUpdate, + WebhookIntegrationCreateResponse, + WebhookIntegrationFilter, + WebhookIntegrationRequest, + WebhookIntegrationResponse, + WebhookIntegrationResponseBody, + WebhookIntegrationResponseMetadata, + WebhookIntegrationResponseResources, + WebhookIntegrationRotateSecretRequest, + WebhookIntegrationSecretResponse, + WebhookIntegrationStats, + WebhookIntegrationUpdate, +) from zenml.models.v2.misc.auth_models import ( OAuthDeviceAuthorizationRequest, OAuthDeviceAuthorizationResponse, @@ -643,6 +657,11 @@ PlatformEventTriggerUpdate.model_rebuild() PlatformEventTriggerResponse.model_rebuild() PlatformEventTriggerResponseBody.model_rebuild() +WebhookIntegrationResponseBody.model_rebuild() +WebhookIntegrationStats.model_rebuild() +WebhookIntegrationResponseMetadata.model_rebuild() +WebhookIntegrationResponseResources.model_rebuild() +WebhookIntegrationResponse.model_rebuild() __all__ = [ @@ -1027,4 +1046,16 @@ "TRIGGER_RETURN_TYPE_UNION", "SourceEntity", "TriggerExecutionInfo", + "WebhookEventStatsUpdate", + "WebhookIntegrationCreateResponse", + "WebhookIntegrationFilter", + "WebhookIntegrationRequest", + "WebhookIntegrationResponse", + "WebhookIntegrationResponseBody", + "WebhookIntegrationResponseMetadata", + "WebhookIntegrationResponseResources", + "WebhookIntegrationRotateSecretRequest", + "WebhookIntegrationSecretResponse", + "WebhookIntegrationStats", + "WebhookIntegrationUpdate", ] diff --git a/src/zenml/models/v2/core/webhook_integration.py b/src/zenml/models/v2/core/webhook_integration.py new file mode 100644 index 00000000000..182e5648465 --- /dev/null +++ b/src/zenml/models/v2/core/webhook_integration.py @@ -0,0 +1,211 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""Models for project-scoped webhook integrations.""" + +from datetime import datetime +from typing import ClassVar + +from pydantic import Field, model_validator + +from zenml.constants import STR_FIELD_MAX_LENGTH +from zenml.enums import WebhookType +from zenml.models.v2.base.base import BaseUpdate, BaseZenModel +from zenml.models.v2.base.filter import EnumFilterOption, StringFilterOption +from zenml.models.v2.base.scoped import ( + ProjectScopedFilter, + ProjectScopedRequest, + ProjectScopedResponse, + ProjectScopedResponseBody, + ProjectScopedResponseMetadata, + ProjectScopedResponseResources, +) +from zenml.utils.secret_utils import ( + NonEmptyPlainSerializedSecretStr, + PlainSerializedSecretStr, +) + + +class WebhookIntegrationRequest(ProjectScopedRequest): + """Request model for creating a webhook integration.""" + + name: str = Field(max_length=STR_FIELD_MAX_LENGTH) + webhook_type: WebhookType + active: bool = True + secret: NonEmptyPlainSerializedSecretStr | None = Field( + default=None, + description="Optional signing secret. A secret is generated when " + "omitted.", + ) + + +class WebhookIntegrationUpdate(BaseUpdate): + """Request model for updating a webhook integration.""" + + name: str | None = Field(default=None, max_length=STR_FIELD_MAX_LENGTH) + active: bool | None = None + + +class WebhookIntegrationResponseBody(ProjectScopedResponseBody): + """Response body for a webhook integration.""" + + webhook_type: WebhookType + active: bool + endpoint_path: str + + +class WebhookIntegrationStats(BaseZenModel): + """Intake statistics for a webhook integration.""" + + received_count: int = 0 + accepted_count: int = 0 + auth_failed_count: int = 0 + invalid_payload_count: int = 0 + last_received_at: datetime | None = None + last_accepted_at: datetime | None = None + last_error_at: datetime | None = None + last_error_summary: str | None = None + + +class WebhookIntegrationResponseMetadata(ProjectScopedResponseMetadata): + """Response metadata for a webhook integration.""" + + stats: WebhookIntegrationStats = Field( + default_factory=WebhookIntegrationStats + ) + + +class WebhookIntegrationResponseResources(ProjectScopedResponseResources): + """Resources associated with a webhook integration.""" + + +class WebhookIntegrationResponse( + ProjectScopedResponse[ + WebhookIntegrationResponseBody, + WebhookIntegrationResponseMetadata, + WebhookIntegrationResponseResources, + ] +): + """Response model for a webhook integration.""" + + name: str = Field(max_length=STR_FIELD_MAX_LENGTH) + + def get_hydrated_version(self) -> "WebhookIntegrationResponse": + """Return the hydrated webhook integration. + + Returns: + The hydrated webhook integration. + """ + from zenml.client import Client + + return Client().zen_store.get_webhook_integration(self.id) + + @property + def webhook_type(self) -> WebhookType: + """Return the webhook provider type. + + Returns: + The webhook provider type. + """ + return self.get_body().webhook_type + + @property + def active(self) -> bool: + """Return whether the integration accepts events. + + Returns: + Whether the integration accepts events. + """ + return self.get_body().active + + @property + def endpoint_path(self) -> str: + """Return the provider-facing event endpoint path. + + Returns: + The provider-facing event endpoint path. + """ + return self.get_body().endpoint_path + + @property + def stats(self) -> WebhookIntegrationStats: + """Return intake statistics for this webhook integration. + + Returns: + Intake statistics for this webhook integration. + """ + return self.get_metadata().stats + + +class WebhookIntegrationFilter(ProjectScopedFilter): + """Filter model for webhook integrations.""" + + name: StringFilterOption = None + webhook_type: EnumFilterOption[WebhookType] = None + API_SINGLE_INPUT_PARAMS: ClassVar[list[str]] = [ + *ProjectScopedFilter.API_SINGLE_INPUT_PARAMS, + "active", + ] + + active: bool | None = None + + +class WebhookIntegrationCreateResponse(BaseZenModel): + """Creation result with a generated secret when applicable.""" + + webhook: WebhookIntegrationResponse + secret: PlainSerializedSecretStr | None = None + + +class WebhookIntegrationRotateSecretRequest(BaseZenModel): + """Request model for rotating a webhook integration secret.""" + + secret: NonEmptyPlainSerializedSecretStr | None = Field( + default=None, + description="Optional direct replacement secret.", + ) + + +class WebhookIntegrationSecretResponse(BaseZenModel): + """One-time response containing a newly active signing secret.""" + + secret: PlainSerializedSecretStr + + +class WebhookEventStatsUpdate(BaseZenModel): + """Atomic intake statistics update.""" + + accepted: bool = False + auth_failed: bool = False + invalid_payload: bool = False + error_summary: str | None = Field(default=None, max_length=4096) + + @model_validator(mode="after") + def validate_single_outcome(self) -> "WebhookEventStatsUpdate": + """Validate that exactly one terminal intake outcome is selected. + + Returns: + The validated statistics update. + + Raises: + ValueError: If the update does not contain exactly one outcome or + an accepted outcome contains an error. + """ + outcomes = (self.accepted, self.auth_failed, self.invalid_payload) + if sum(outcomes) != 1: + raise ValueError("Exactly one webhook intake outcome is required.") + if self.accepted and self.error_summary is not None: + raise ValueError( + "Accepted webhook events cannot include an error." + ) + return self diff --git a/src/zenml/utils/secret_utils.py b/src/zenml/utils/secret_utils.py index 744675376d0..c5ae739fe60 100644 --- a/src/zenml/utils/secret_utils.py +++ b/src/zenml/utils/secret_utils.py @@ -16,7 +16,7 @@ import re from typing import TYPE_CHECKING, Any, List, NamedTuple, Union -from pydantic import Field, PlainSerializer, SecretStr +from pydantic import AfterValidator, Field, PlainSerializer, SecretStr from typing_extensions import Annotated from zenml.logger import get_logger @@ -39,6 +39,33 @@ ), ] + +def _validate_non_empty_secret(secret: SecretStr) -> SecretStr: + """Validate that a secret contains non-whitespace content. + + Args: + secret: The secret to validate. + + Returns: + The validated secret. + + Raises: + ValueError: If the secret is empty or contains only whitespace. + """ + if not secret.get_secret_value().strip(): + raise ValueError("Secret must not be empty.") + return secret + + +NonEmptyPlainSerializedSecretStr = Annotated[ + SecretStr, + AfterValidator(_validate_non_empty_secret), + PlainSerializer( + lambda v: v.get_secret_value() if v is not None else None, + when_used="json", + ), +] + logger = get_logger(__name__) diff --git a/src/zenml/webhooks/__init__.py b/src/zenml/webhooks/__init__.py new file mode 100644 index 00000000000..50cf05e1c68 --- /dev/null +++ b/src/zenml/webhooks/__init__.py @@ -0,0 +1,28 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""Webhook provider authentication and payload validation.""" + +from zenml.webhooks.adapters import ( + WebhookAuthenticationError, + WebhookEvent, + WebhookPayloadError, + get_webhook_adapter, +) + +__all__ = [ + "WebhookAuthenticationError", + "WebhookEvent", + "WebhookPayloadError", + "get_webhook_adapter", +] diff --git a/src/zenml/webhooks/adapters.py b/src/zenml/webhooks/adapters.py new file mode 100644 index 00000000000..d06c87b630a --- /dev/null +++ b/src/zenml/webhooks/adapters.py @@ -0,0 +1,283 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""Provider adapters for webhook authentication and basic validation.""" + +import hashlib +import hmac +import json +from abc import ABC, abstractmethod +from collections.abc import Mapping +from typing import Any + +from pydantic import BaseModel + +from zenml.enums import WebhookType +from zenml.exceptions import CredentialsNotValid + + +class WebhookAuthenticationError(CredentialsNotValid): + """Raised when a webhook request cannot be authenticated.""" + + +class WebhookPayloadError(ValueError): + """Raised when a webhook payload fails fundamental validation.""" + + +class WebhookEvent(BaseModel): + """Authenticated provider event produced by an intake adapter.""" + + webhook_type: WebhookType + event_type: str + delivery_id: str | None = None + payload: dict[str, Any] + + +class BaseWebhookAdapter(ABC): + """Base class for provider-specific webhook adapters.""" + + webhook_type: WebhookType + + def validate( + self, body: bytes, headers: Mapping[str, str], secret: str + ) -> WebhookEvent: + """Authenticate and structurally validate a provider event. + + Args: + body: The exact raw request body. + headers: The request headers. + secret: The integration authentication secret. + + Returns: + The authenticated webhook event. + + Raises: + WebhookAuthenticationError: If authentication fails. + WebhookPayloadError: If the event metadata or payload is invalid. + """ # noqa: DOC502 + self.authenticate(body=body, headers=headers, secret=secret) + return self.parse(body=body, headers=headers) + + def parse(self, body: bytes, headers: Mapping[str, str]) -> WebhookEvent: + """Structurally validate and parse a provider event. + + Args: + body: The exact raw request body. + headers: The request headers. + + Returns: + The parsed webhook event. + + Raises: + WebhookPayloadError: If the event metadata or payload is invalid. + """ + try: + payload = json.loads(body) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise WebhookPayloadError( + "Request body must be valid JSON." + ) from error + if not isinstance(payload, dict): + raise WebhookPayloadError( + "Request body must contain a top-level JSON object." + ) + event_type = self.get_event_type(payload=payload, headers=headers) + return WebhookEvent( + webhook_type=self.webhook_type, + event_type=event_type, + delivery_id=self.get_delivery_id(payload=payload, headers=headers), + payload=payload, + ) + + @abstractmethod + def get_event_type( + self, payload: dict[str, Any], headers: Mapping[str, str] + ) -> str: + """Extract the provider event type. + + Args: + payload: The parsed JSON object. + headers: The request headers. + + Returns: + The provider event type. + + Raises: + WebhookPayloadError: If the event type is missing or invalid. + """ + + def get_delivery_id( + self, payload: dict[str, Any], headers: Mapping[str, str] + ) -> str | None: + """Extract the optional provider delivery ID. + + Args: + payload: The parsed JSON object. + headers: The request headers. + + Returns: + The provider delivery ID, if available. + """ + return None + + @abstractmethod + def authenticate( + self, body: bytes, headers: Mapping[str, str], secret: str + ) -> None: + """Authenticate the raw request body. + + Args: + body: The exact raw request body. + headers: The request headers. + secret: The integration authentication secret. + """ + + +class HMACSHA256WebhookAdapter(BaseWebhookAdapter): + """Webhook adapter using a sha256-prefixed HMAC signature.""" + + signature_header: str + + def authenticate( + self, body: bytes, headers: Mapping[str, str], secret: str + ) -> None: + """Validate an HMAC-SHA256 signature over the raw request body. + + Args: + body: The exact raw request body. + headers: The request headers. + secret: The integration signing secret. + + Raises: + WebhookAuthenticationError: If the signature is missing, + malformed, or invalid. + """ + signature = headers.get(self.signature_header) + if not signature or not signature.startswith("sha256="): + raise WebhookAuthenticationError( + f"Missing or malformed {self.signature_header} header." + ) + expected = ( + "sha256=" + + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + ) + if not hmac.compare_digest(signature, expected): + raise WebhookAuthenticationError("Invalid webhook signature.") + + +class GitHubWebhookAdapter(HMACSHA256WebhookAdapter): + """GitHub webhook adapter.""" + + webhook_type = WebhookType.GITHUB + signature_header = "x-hub-signature-256" + event_header = "x-github-event" + delivery_header = "x-github-delivery" + + def get_event_type( + self, payload: dict[str, Any], headers: Mapping[str, str] + ) -> str: + """Extract the GitHub event type from its request header. + + Args: + payload: The parsed JSON object. + headers: The request headers. + + Returns: + The GitHub event type. + + Raises: + WebhookPayloadError: If the event type header is missing. + """ + event_type = headers.get(self.event_header) + if not event_type: + raise WebhookPayloadError( + f"Missing required {self.event_header} header." + ) + return event_type + + def get_delivery_id( + self, payload: dict[str, Any], headers: Mapping[str, str] + ) -> str | None: + """Extract the optional GitHub delivery ID. + + Args: + payload: The parsed JSON object. + headers: The request headers. + + Returns: + The GitHub delivery ID, if available. + """ + return headers.get(self.delivery_header) + + +class CustomWebhookAdapter(HMACSHA256WebhookAdapter): + """ZenML custom webhook adapter.""" + + webhook_type = WebhookType.CUSTOM + signature_header = "x-zenml-signature-256" + event_header = "x-zenml-event" + delivery_header = "x-zenml-delivery" + + def get_event_type( + self, payload: dict[str, Any], headers: Mapping[str, str] + ) -> str: + """Extract the custom event type from its request header. + + Args: + payload: The parsed JSON object. + headers: The request headers. + + Returns: + The custom event type. + + Raises: + WebhookPayloadError: If the event type header is missing. + """ + event_type = headers.get(self.event_header) + if not event_type: + raise WebhookPayloadError( + f"Missing required {self.event_header} header." + ) + return event_type + + def get_delivery_id( + self, payload: dict[str, Any], headers: Mapping[str, str] + ) -> str | None: + """Extract the optional custom delivery ID. + + Args: + payload: The parsed JSON object. + headers: The request headers. + + Returns: + The custom delivery ID, if available. + """ + return headers.get(self.delivery_header) + + +_ADAPTERS: dict[WebhookType, BaseWebhookAdapter] = { + WebhookType.GITHUB: GitHubWebhookAdapter(), + WebhookType.CUSTOM: CustomWebhookAdapter(), +} + + +def get_webhook_adapter(webhook_type: WebhookType) -> BaseWebhookAdapter: + """Return the registered adapter for a webhook provider type. + + Args: + webhook_type: The webhook provider type. + + Returns: + The registered provider adapter. + """ + return _ADAPTERS[webhook_type] diff --git a/src/zenml/zen_server/rbac/models.py b/src/zenml/zen_server/rbac/models.py index 502e0dbd1e6..afd63843e55 100644 --- a/src/zenml/zen_server/rbac/models.py +++ b/src/zenml/zen_server/rbac/models.py @@ -78,6 +78,7 @@ class ResourceType(StrEnum): # Deactivated for now # USER = "user" TRIGGER = "trigger" + WEBHOOK_INTEGRATION = "webhook_integration" def is_project_scoped(self) -> bool: """Check if a resource type is project scoped. diff --git a/src/zenml/zen_server/rbac/utils.py b/src/zenml/zen_server/rbac/utils.py index 90eefbb0cba..e7fc0a74420 100644 --- a/src/zenml/zen_server/rbac/utils.py +++ b/src/zenml/zen_server/rbac/utils.py @@ -522,6 +522,8 @@ def get_resource_type_for_model( TagRequest, TagResponse, TriggerRequest, + WebhookIntegrationRequest, + WebhookIntegrationResponse, ) mapping: Dict[ @@ -581,6 +583,8 @@ def get_resource_type_for_model( ScheduleTriggerRequest: ResourceType.TRIGGER, ScheduleTriggerResponse: ResourceType.TRIGGER, TriggerRequest: ResourceType.TRIGGER, + WebhookIntegrationRequest: ResourceType.WEBHOOK_INTEGRATION, + WebhookIntegrationResponse: ResourceType.WEBHOOK_INTEGRATION, } return mapping.get(type(model)) @@ -714,6 +718,7 @@ def _get_resource_type_schema_mapping() -> Dict[ TagSchema, TriggerSchema, UserSchema, + WebhookIntegrationSchema, ) return { @@ -742,6 +747,7 @@ def _get_resource_type_schema_mapping() -> Dict[ ResourceType.SCHEDULE: ScheduleSchema, # ResourceType.USER: UserSchema, ResourceType.TRIGGER: TriggerSchema, + ResourceType.WEBHOOK_INTEGRATION: WebhookIntegrationSchema, ResourceType.RESOURCE_POOL_SUBJECT_POLICY: ResourcePoolSubjectPolicySchema, } diff --git a/src/zenml/zen_server/routers/webhook_integration_endpoints.py b/src/zenml/zen_server/routers/webhook_integration_endpoints.py new file mode 100644 index 00000000000..1806703f145 --- /dev/null +++ b/src/zenml/zen_server/routers/webhook_integration_endpoints.py @@ -0,0 +1,321 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""Management and public intake endpoints for webhook integrations.""" + +from uuid import UUID + +from fastapi import ( + APIRouter, + Depends, + HTTPException, + Request, + Security, + status, +) +from fastapi.responses import Response +from starlette.concurrency import run_in_threadpool +from starlette.datastructures import Headers + +from zenml.constants import ( + API, + VERSION_1, + WEBHOOKS, +) +from zenml.enums import WebhookType +from zenml.models import ( + Page, + WebhookEventStatsUpdate, + WebhookIntegrationCreateResponse, + WebhookIntegrationFilter, + WebhookIntegrationRequest, + WebhookIntegrationResponse, + WebhookIntegrationRotateSecretRequest, + WebhookIntegrationSecretResponse, + WebhookIntegrationUpdate, +) +from zenml.webhooks import ( + WebhookAuthenticationError, + WebhookPayloadError, + get_webhook_adapter, +) +from zenml.zen_server.auth import AuthContext, authorize +from zenml.zen_server.exceptions import error_response +from zenml.zen_server.rbac.endpoint_utils import ( + verify_permissions_and_delete_entity, + verify_permissions_and_get_entity, + verify_permissions_and_list_entities, +) +from zenml.zen_server.rbac.models import Action, ResourceType +from zenml.zen_server.rbac.utils import ( + dehydrate_response_model, + verify_permission_for_model, +) +from zenml.zen_server.utils import ( + async_fastapi_endpoint_wrapper, + async_handle_endpoint_errors, + make_dependable, + zen_store, +) + +management_router = APIRouter( + prefix=API + VERSION_1 + WEBHOOKS, + tags=["webhooks"], + responses={401: error_response, 403: error_response}, +) + +intake_router = APIRouter( + prefix=API + VERSION_1 + WEBHOOKS, + tags=["webhooks"], +) + + +@management_router.post("") +@async_fastapi_endpoint_wrapper +def create_webhook( + integration: WebhookIntegrationRequest, + _: AuthContext = Security(authorize), +) -> WebhookIntegrationCreateResponse: + """Create a project-scoped webhook. + + Args: + integration: The webhook creation request. + + Returns: + The created integration and any generated signing secret. + """ + verify_permission_for_model(model=integration, action=Action.CREATE) + result = zen_store().create_webhook_integration(integration) + return dehydrate_response_model(result) + + +@management_router.get("") +@async_fastapi_endpoint_wrapper +def list_webhooks( + filter_model: WebhookIntegrationFilter = Depends( + make_dependable(WebhookIntegrationFilter) + ), + hydrate: bool = False, + _: AuthContext = Security(authorize), +) -> Page[WebhookIntegrationResponse]: + """List webhooks. + + Args: + filter_model: The webhook filters. + hydrate: Whether to include intake statistics. + + Returns: + A page of webhooks. + """ + return verify_permissions_and_list_entities( + filter_model=filter_model, + resource_type=ResourceType.WEBHOOK_INTEGRATION, + list_method=zen_store().list_webhook_integrations, + hydrate=hydrate, + ) + + +@management_router.get("/{webhook_id}") +@async_fastapi_endpoint_wrapper +def get_webhook( + webhook_id: UUID, + hydrate: bool = True, + _: AuthContext = Security(authorize), +) -> WebhookIntegrationResponse: + """Get a webhook. + + Args: + webhook_id: The webhook ID. + hydrate: Whether to include intake statistics. + + Returns: + The webhook. + """ + return verify_permissions_and_get_entity( + id=webhook_id, + get_method=zen_store().get_webhook_integration, + hydrate=hydrate, + ) + + +@management_router.put("/{webhook_id}") +@async_fastapi_endpoint_wrapper +def update_webhook( + webhook_id: UUID, + update: WebhookIntegrationUpdate, + _: AuthContext = Security(authorize), +) -> WebhookIntegrationResponse: + """Update a webhook. + + Args: + webhook_id: The webhook ID. + update: The webhook update. + + Returns: + The updated webhook. + """ + integration = zen_store().get_webhook_integration( + webhook_id, hydrate=False + ) + verify_permission_for_model(model=integration, action=Action.UPDATE) + updated_integration = zen_store().update_webhook_integration( + integration_id=integration.id, + update=update, + ) + return dehydrate_response_model(updated_integration) + + +@management_router.delete("/{webhook_id}") +@async_fastapi_endpoint_wrapper +def delete_webhook( + webhook_id: UUID, + _: AuthContext = Security(authorize), +) -> None: + """Delete a webhook and its signing secret. + + Args: + webhook_id: The webhook ID. + """ + verify_permissions_and_delete_entity( + id=webhook_id, + get_method=zen_store().get_webhook_integration, + delete_method=zen_store().delete_webhook_integration, + ) + + +@management_router.put("/{webhook_id}/secret") +@async_fastapi_endpoint_wrapper +def rotate_webhook_secret( + webhook_id: UUID, + request: WebhookIntegrationRotateSecretRequest, + _: AuthContext = Security(authorize), +) -> WebhookIntegrationSecretResponse: + """Rotate a webhook signing secret. + + Args: + webhook_id: The webhook ID. + request: The secret rotation request. + + Returns: + The newly active signing secret. + """ + integration = zen_store().get_webhook_integration(webhook_id) + verify_permission_for_model(model=integration, action=Action.UPDATE) + return zen_store().rotate_webhook_integration_secret( + integration_id=webhook_id, request=request + ) + + +@intake_router.post( + "/{webhook_type}/{webhook_id}/events", + status_code=status.HTTP_202_ACCEPTED, +) +@async_handle_endpoint_errors +async def receive_webhook_event( + webhook_type: WebhookType, + webhook_id: UUID, + request: Request, +) -> Response: + """Authenticate and accept a provider webhook event. + + Args: + webhook_type: The provider type from the public endpoint path. + webhook_id: The webhook ID. + request: The raw HTTP request. + + Returns: + An empty accepted response. + """ + body = await request.body() + return await run_in_threadpool( + _receive_webhook_event, + webhook_type=webhook_type, + integration_id=webhook_id, + body=body, + headers=request.headers, + ) + + +def _receive_webhook_event( + webhook_type: WebhookType, + integration_id: UUID, + body: bytes, + headers: Headers, +) -> Response: + """Synchronously authenticate and accept a webhook event. + + Args: + webhook_type: The provider type from the public endpoint path. + integration_id: The webhook ID. + body: The raw request body. + headers: The request headers. + + Returns: + An empty accepted response. + + Raises: + HTTPException: If the integration cannot accept the event or the + request fails authentication or payload validation. + """ + try: + stored_type, active, secret_id = zen_store().get_webhook_intake_config( + integration_id + ) + except KeyError as error: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) from error + + if stored_type != webhook_type: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) + + secret = zen_store().get_webhook_secret(secret_id) + adapter = get_webhook_adapter(webhook_type) + + try: + adapter.authenticate(body=body, headers=headers, secret=secret) + except WebhookAuthenticationError as error: + if active: + zen_store().record_webhook_event( + integration_id, + WebhookEventStatsUpdate( + auth_failed=True, error_summary=str(error) + ), + ) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid webhook authentication.", + ) from error + + if not active: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Webhook integration is inactive.", + ) + + try: + adapter.parse(body=body, headers=headers) + except WebhookPayloadError as error: + zen_store().record_webhook_event( + integration_id, + WebhookEventStatsUpdate( + invalid_payload=True, error_summary=str(error) + ), + ) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(error), + ) from error + + zen_store().record_webhook_event( + integration_id, WebhookEventStatsUpdate(accepted=True) + ) + return Response(status_code=status.HTTP_202_ACCEPTED) diff --git a/src/zenml/zen_server/zen_server_api.py b/src/zenml/zen_server/zen_server_api.py index c908f862ce7..fca6dd838ce 100644 --- a/src/zenml/zen_server/zen_server_api.py +++ b/src/zenml/zen_server/zen_server_api.py @@ -95,6 +95,7 @@ tags_endpoints, trigger_endpoints, users_endpoints, + webhook_integration_endpoints, ) from zenml.zen_server.secure_headers import ( initialize_secure_headers, @@ -349,6 +350,8 @@ async def dashboard(request: Request) -> Any: app.include_router(resource_pool_subject_policies_endpoints.router) app.include_router(resource_requests_endpoints.router) app.include_router(trigger_endpoints.router) +app.include_router(webhook_integration_endpoints.management_router) +app.include_router(webhook_integration_endpoints.intake_router) # When the auth scheme is set to EXTERNAL, users cannot be managed via the # API. diff --git a/src/zenml/zen_stores/migrations/versions/7c0d9e4a1b2f_add_webhook_integrations.py b/src/zenml/zen_stores/migrations/versions/7c0d9e4a1b2f_add_webhook_integrations.py new file mode 100644 index 00000000000..97be2bea021 --- /dev/null +++ b/src/zenml/zen_stores/migrations/versions/7c0d9e4a1b2f_add_webhook_integrations.py @@ -0,0 +1,122 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""Add webhook integrations [7c0d9e4a1b2f]. + +Revision ID: 7c0d9e4a1b2f +Revises: b6f2a8d9c3e1 +Create Date: 2026-07-07 + +""" + +import sqlalchemy as sa +import sqlmodel +from alembic import op + +revision = "7c0d9e4a1b2f" +down_revision = "b6f2a8d9c3e1" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + """Create the webhook integration table.""" + op.create_table( + "webhook_integration", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("created", sa.DateTime(), nullable=False), + sa.Column("updated", sa.DateTime(), nullable=False), + sa.Column("name", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("project_id", sa.Uuid(), nullable=False), + sa.Column("user_id", sa.Uuid(), nullable=True), + sa.Column("secret_id", sa.Uuid(), nullable=False), + sa.Column( + "webhook_type", sqlmodel.sql.sqltypes.AutoString(), nullable=False + ), + sa.Column("active", sa.Boolean(), nullable=False), + sa.ForeignKeyConstraint( + ["project_id"], + ["project.id"], + name="fk_webhook_integration_project_id_project", + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["user_id"], + ["user.id"], + name="fk_webhook_integration_user_id_user", + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["secret_id"], + ["secret.id"], + name="fk_webhook_integration_secret_id_secret", + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "project_id", + "name", + name="unique_webhook_integration_name_in_project", + ), + ) + op.create_index( + "ix_webhook_integration_webhook_type", + "webhook_integration", + ["webhook_type"], + unique=False, + ) + op.create_table( + "webhook_integration_stats", + sa.Column("webhook_id", sa.Uuid(), nullable=False), + sa.Column( + "received_count", sa.Integer(), nullable=False, server_default="0" + ), + sa.Column( + "accepted_count", sa.Integer(), nullable=False, server_default="0" + ), + sa.Column( + "auth_failed_count", + sa.Integer(), + nullable=False, + server_default="0", + ), + sa.Column( + "invalid_payload_count", + sa.Integer(), + nullable=False, + server_default="0", + ), + sa.Column("last_received_at", sa.DateTime(), nullable=True), + sa.Column("last_accepted_at", sa.DateTime(), nullable=True), + sa.Column("last_error_at", sa.DateTime(), nullable=True), + sa.Column("last_error_summary", sa.TEXT(), nullable=True), + sa.ForeignKeyConstraint( + ["webhook_id"], + ["webhook_integration.id"], + name=( + "fk_webhook_integration_stats_webhook_id_webhook_integration" + ), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("webhook_id"), + ) + + +def downgrade() -> None: + """Drop the webhook integration table.""" + op.drop_table("webhook_integration_stats") + op.drop_index( + "ix_webhook_integration_webhook_type", + table_name="webhook_integration", + ) + op.drop_table("webhook_integration") diff --git a/src/zenml/zen_stores/rest_zen_store.py b/src/zenml/zen_stores/rest_zen_store.py index 80365b8d8b4..1b715c951db 100644 --- a/src/zenml/zen_stores/rest_zen_store.py +++ b/src/zenml/zen_stores/rest_zen_store.py @@ -126,6 +126,7 @@ TRIGGERS, USERS, VERSION_1, + WEBHOOKS, ZENML_PRO_API_KEY_PREFIX, ) from zenml.enums import ( @@ -307,6 +308,13 @@ UserRequest, UserResponse, UserUpdate, + WebhookIntegrationCreateResponse, + WebhookIntegrationFilter, + WebhookIntegrationRequest, + WebhookIntegrationResponse, + WebhookIntegrationRotateSecretRequest, + WebhookIntegrationSecretResponse, + WebhookIntegrationUpdate, ) from zenml.service_connectors.service_connector_registry import ( service_connector_registry, @@ -2577,6 +2585,107 @@ def create_run_metadata(self, run_metadata: RunMetadataRequest) -> None: """ self.post(RUN_METADATA, body=run_metadata) + # ------------------------ Webhook integrations ----------------------- + + def create_webhook_integration( + self, integration: WebhookIntegrationRequest + ) -> WebhookIntegrationCreateResponse: + """Create a webhook integration. + + Args: + integration: The webhook integration creation request. + + Returns: + The created integration and any generated signing secret. + """ + response = self.post(WEBHOOKS, body=integration) + return WebhookIntegrationCreateResponse.model_validate(response) + + def get_webhook_integration( + self, integration_id: UUID, hydrate: bool = True + ) -> WebhookIntegrationResponse: + """Get a webhook integration. + + Args: + integration_id: The webhook integration ID. + hydrate: Whether to include intake statistics. + + Returns: + The webhook integration. + """ + response = self.get( + f"{WEBHOOKS}/{integration_id}", + params={"hydrate": hydrate}, + ) + return WebhookIntegrationResponse.model_validate(response) + + def list_webhook_integrations( + self, + filter_model: WebhookIntegrationFilter, + hydrate: bool = False, + ) -> Page[WebhookIntegrationResponse]: + """List webhook integrations. + + Args: + filter_model: The webhook integration filters. + hydrate: Whether to include intake statistics. + + Returns: + A page of webhook integrations. + """ + response = self.get( + WEBHOOKS, + params={ + "hydrate": hydrate, + **filter_model.model_dump(exclude_none=True), + }, + ) + return Page[WebhookIntegrationResponse].model_validate(response) + + def update_webhook_integration( + self, + integration_id: UUID, + update: WebhookIntegrationUpdate, + ) -> WebhookIntegrationResponse: + """Update a webhook integration. + + Args: + integration_id: The webhook integration ID. + update: The webhook integration update. + + Returns: + The updated webhook integration. + """ + response = self.put(f"{WEBHOOKS}/{integration_id}", body=update) + return WebhookIntegrationResponse.model_validate(response) + + def delete_webhook_integration(self, integration_id: UUID) -> None: + """Delete a webhook integration. + + Args: + integration_id: The webhook integration ID. + """ + self.delete(f"{WEBHOOKS}/{integration_id}") + + def rotate_webhook_integration_secret( + self, + integration_id: UUID, + request: WebhookIntegrationRotateSecretRequest, + ) -> WebhookIntegrationSecretResponse: + """Rotate a webhook integration signing secret. + + Args: + integration_id: The webhook integration ID. + request: The secret rotation request. + + Returns: + The newly active signing secret. + """ + response = self.put( + f"{WEBHOOKS}/{integration_id}/secret", body=request + ) + return WebhookIntegrationSecretResponse.model_validate(response) + # ----------------------------- Triggers ------------------------------ def create_trigger( diff --git a/src/zenml/zen_stores/schemas/__init__.py b/src/zenml/zen_stores/schemas/__init__.py index c6a8f52e309..3055cc345cc 100644 --- a/src/zenml/zen_stores/schemas/__init__.py +++ b/src/zenml/zen_stores/schemas/__init__.py @@ -110,6 +110,10 @@ ) from zenml.zen_stores.schemas.trigger_schemas import TriggerSchema from zenml.zen_stores.schemas.user_schemas import UserSchema +from zenml.zen_stores.schemas.webhook_integration_schemas import ( + WebhookIntegrationSchema, + WebhookIntegrationStatsSchema, +) __all__ = [ "APIKeySchema", @@ -172,4 +176,6 @@ "TriggerSchema", "TriggerSnapshotSchema", "TriggerExecutionSchema", + "WebhookIntegrationSchema", + "WebhookIntegrationStatsSchema", ] diff --git a/src/zenml/zen_stores/schemas/project_schemas.py b/src/zenml/zen_stores/schemas/project_schemas.py index 144f90ca15c..97603939a10 100644 --- a/src/zenml/zen_stores/schemas/project_schemas.py +++ b/src/zenml/zen_stores/schemas/project_schemas.py @@ -46,6 +46,7 @@ ServiceSchema, StepRunSchema, TriggerSchema, + WebhookIntegrationSchema, ) @@ -132,6 +133,10 @@ class ProjectSchema(NamedSchema, table=True): back_populates="project", sa_relationship_kwargs={"cascade": "delete"}, ) + webhook_integrations: list["WebhookIntegrationSchema"] = Relationship( + back_populates="project", + sa_relationship_kwargs={"cascade": "delete"}, + ) @classmethod def from_request(cls, project: ProjectRequest) -> "ProjectSchema": diff --git a/src/zenml/zen_stores/schemas/webhook_integration_schemas.py b/src/zenml/zen_stores/schemas/webhook_integration_schemas.py new file mode 100644 index 00000000000..22824c995d4 --- /dev/null +++ b/src/zenml/zen_stores/schemas/webhook_integration_schemas.py @@ -0,0 +1,249 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""SQL schemas for webhook integrations.""" + +from datetime import datetime +from typing import Any, Sequence +from uuid import UUID + +from sqlalchemy import TEXT, Column, UniqueConstraint +from sqlalchemy.orm import joinedload, selectinload +from sqlalchemy.sql.base import ExecutableOption +from sqlmodel import Field, Relationship, SQLModel + +from zenml.constants import API, VERSION_1, WEBHOOKS +from zenml.models import ( + WebhookIntegrationRequest, + WebhookIntegrationResponse, + WebhookIntegrationResponseBody, + WebhookIntegrationResponseMetadata, + WebhookIntegrationResponseResources, + WebhookIntegrationStats, + WebhookIntegrationUpdate, +) +from zenml.utils.time_utils import utc_now +from zenml.zen_stores.schemas.base_schemas import NamedSchema +from zenml.zen_stores.schemas.project_schemas import ProjectSchema +from zenml.zen_stores.schemas.schema_utils import ( + build_foreign_key_field, + build_index, +) +from zenml.zen_stores.schemas.secret_schemas import SecretSchema +from zenml.zen_stores.schemas.user_schemas import UserSchema +from zenml.zen_stores.schemas.utils import jl_arg + + +class WebhookIntegrationSchema(NamedSchema, table=True): + """SQL schema for a project-scoped webhook integration.""" + + __tablename__ = "webhook_integration" + __table_args__ = ( + UniqueConstraint( + "project_id", + "name", + name="unique_webhook_integration_name_in_project", + ), + build_index( + table_name=__tablename__, + column_names=["webhook_type"], + ), + ) + + project_id: UUID = build_foreign_key_field( + source=__tablename__, + target=ProjectSchema.__tablename__, + source_column="project_id", + target_column="id", + ondelete="CASCADE", + nullable=False, + ) + project: "ProjectSchema" = Relationship( + back_populates="webhook_integrations" + ) + user_id: UUID | None = build_foreign_key_field( + source=__tablename__, + target=UserSchema.__tablename__, + source_column="user_id", + target_column="id", + ondelete="SET NULL", + nullable=True, + ) + user: UserSchema | None = Relationship() + secret_id: UUID = build_foreign_key_field( + source=__tablename__, + target=SecretSchema.__tablename__, + source_column="secret_id", + target_column="id", + ondelete="RESTRICT", + nullable=False, + ) + webhook_type: str + active: bool = Field(default=True) + stats: "WebhookIntegrationStatsSchema" = Relationship( + back_populates="webhook", + sa_relationship_kwargs={ + "cascade": "all, delete-orphan", + "single_parent": True, + "uselist": False, + }, + ) + + @classmethod + def get_query_options( + cls, + include_metadata: bool = False, + include_resources: bool = False, + **kwargs: Any, + ) -> Sequence[ExecutableOption]: + """Get query options for webhook integrations. + + Args: + include_metadata: Whether statistics will be included. + include_resources: Whether related resources will be included. + **kwargs: Additional query option arguments. + + Returns: + Query options for the requested response shape. + """ + options: list[ExecutableOption] = [] + if include_metadata: + options.append(selectinload(jl_arg(cls.stats))) + if include_resources: + options.append(joinedload(jl_arg(cls.user))) + return options + + @classmethod + def from_request( + cls, request: WebhookIntegrationRequest, secret_id: UUID + ) -> "WebhookIntegrationSchema": + """Create a schema from a request. + + Args: + request: The webhook integration creation request. + secret_id: The internal signing secret ID. + + Returns: + The created webhook integration schema. + """ + return cls( + name=request.name, + project_id=request.project, + user_id=request.user, + webhook_type=request.webhook_type.value, + active=request.active, + secret_id=secret_id, + ) + + def to_model( + self, + include_metadata: bool = False, + include_resources: bool = False, + **kwargs: Any, + ) -> WebhookIntegrationResponse: + """Convert the schema to a response model. + + Args: + include_metadata: Whether to include intake statistics. + include_resources: Whether to include associated resources. + **kwargs: Additional conversion options. + + Returns: + The webhook integration response. + """ + metadata = None + if include_metadata: + metadata = WebhookIntegrationResponseMetadata( + stats=self.stats.to_model() + ) + + resources = None + if include_resources: + resources = WebhookIntegrationResponseResources( + user=self.user.to_model() if self.user else None, + ) + + return WebhookIntegrationResponse( + id=self.id, + name=self.name, + body=WebhookIntegrationResponseBody( + user_id=self.user_id, + project_id=self.project_id, + created=self.created, + updated=self.updated, + webhook_type=self.webhook_type, + active=self.active, + endpoint_path=( + f"{API}{VERSION_1}{WEBHOOKS}/{self.webhook_type}/" + f"{self.id}/events" + ), + ), + metadata=metadata, + resources=resources, + ) + + def update( + self, update: WebhookIntegrationUpdate + ) -> "WebhookIntegrationSchema": + """Apply a webhook integration update. + + Args: + update: The webhook integration update. + + Returns: + The updated webhook integration schema. + """ + if update.name is not None: + self.name = update.name + if update.active is not None: + self.active = update.active + self.updated = utc_now() + return self + + +class WebhookIntegrationStatsSchema(SQLModel, table=True): + """SQL schema for webhook intake statistics.""" + + __tablename__ = "webhook_integration_stats" + + webhook_id: UUID = build_foreign_key_field( + source=__tablename__, + target=WebhookIntegrationSchema.__tablename__, + source_column="webhook_id", + target_column="id", + ondelete="CASCADE", + nullable=False, + primary_key=True, + ) + received_count: int = 0 + accepted_count: int = 0 + auth_failed_count: int = 0 + invalid_payload_count: int = 0 + last_received_at: datetime | None = None + last_accepted_at: datetime | None = None + last_error_at: datetime | None = None + last_error_summary: str | None = Field( + default=None, sa_column=Column(TEXT, nullable=True) + ) + + webhook: WebhookIntegrationSchema = Relationship(back_populates="stats") + + def to_model(self) -> WebhookIntegrationStats: + """Convert persisted statistics to their domain model. + + Returns: + The webhook intake statistics. + """ + return WebhookIntegrationStats.model_validate( + self, from_attributes=True + ) diff --git a/src/zenml/zen_stores/sql_zen_store.py b/src/zenml/zen_stores/sql_zen_store.py index d7a4d49ac70..e2edb74f4e0 100644 --- a/src/zenml/zen_stores/sql_zen_store.py +++ b/src/zenml/zen_stores/sql_zen_store.py @@ -43,6 +43,7 @@ import os import random import re +import secrets import sys import time import uuid @@ -180,6 +181,7 @@ StoreType, TaggableResourceTypes, VisualizationResourceTypes, + WebhookType, ) from zenml.exceptions import ( AuthorizationException, @@ -374,6 +376,14 @@ UserResponse, UserScopedRequest, UserUpdate, + WebhookEventStatsUpdate, + WebhookIntegrationCreateResponse, + WebhookIntegrationFilter, + WebhookIntegrationRequest, + WebhookIntegrationResponse, + WebhookIntegrationRotateSecretRequest, + WebhookIntegrationSecretResponse, + WebhookIntegrationUpdate, ) from zenml.service_connectors.service_connector_registry import ( service_connector_registry, @@ -457,6 +467,8 @@ TriggerSchema, TriggerSnapshotSchema, UserSchema, + WebhookIntegrationSchema, + WebhookIntegrationStatsSchema, ) from zenml.zen_stores.schemas.artifact_visualization_schemas import ( ArtifactVisualizationSchema, @@ -498,6 +510,8 @@ logger = get_logger(__name__) +_WEBHOOK_SECRET_VALUE_KEY = "secret" + ZENML_SQLITE_DB_FILENAME = "zenml.db" @@ -8111,6 +8125,313 @@ def create_run_metadata(self, run_metadata: RunMetadataRequest) -> None: session.commit() return None + # -------------------- Webhook integrations --------------------- + + def _create_webhook_integration_secret( + self, secret: str, session: Session + ) -> UUID: + """Create an internal secret containing a webhook signing secret. + + Args: + secret: The signing secret value. + session: The active database session. + + Returns: + The internal secret ID. + """ + return self._create_secret_schema( + SecretRequest( + user=self._get_active_user(session=session).id, + name=f"webhook-{uuid.uuid4().hex}", + private=False, + values={_WEBHOOK_SECRET_VALUE_KEY: secret}, + ), + session=session, + internal=True, + ).id + + def create_webhook_integration( + self, integration: WebhookIntegrationRequest + ) -> WebhookIntegrationCreateResponse: + """Create a webhook integration and its internal signing secret. + + Args: + integration: The webhook integration creation request. + + Returns: + The created integration and any generated signing secret. + """ # noqa: DOC501, DOC503 + generated_secret = integration.secret is None + if integration.secret is None: + secret = secrets.token_urlsafe(32) + else: + secret = integration.secret.get_secret_value() + with Session(self.engine) as session: + self._set_request_user_id( + request_model=integration, session=session + ) + self._verify_name_uniqueness( + resource=integration, + schema=WebhookIntegrationSchema, + session=session, + ) + secret_id = self._create_webhook_integration_secret( + secret=secret, + session=session, + ) + try: + schema = WebhookIntegrationSchema.from_request( + request=integration, secret_id=secret_id + ) + stats_schema = WebhookIntegrationStatsSchema( + webhook_id=schema.id + ) + session.add(schema) + session.add(stats_schema) + session.commit() + except Exception: + session.rollback() + self._delete_secret_schema( + secret_id=secret_id, session=session + ) + raise + return WebhookIntegrationCreateResponse( + webhook=schema.to_model( + include_metadata=True, include_resources=True + ), + secret=( + PlainSerializedSecretStr(secret) + if generated_secret + else None + ), + ) + + def get_webhook_integration( + self, integration_id: UUID, hydrate: bool = True + ) -> WebhookIntegrationResponse: + """Get a webhook integration. + + Args: + integration_id: The webhook integration ID. + hydrate: Whether to include intake statistics. + + Returns: + The webhook integration. + """ + with Session(self.engine) as session: + query_options = WebhookIntegrationSchema.get_query_options( + include_metadata=hydrate, include_resources=True + ) + schema = self._get_schema_by_id( + resource_id=integration_id, + schema_class=WebhookIntegrationSchema, + session=session, + query_options=query_options, + ) + return schema.to_model( + include_metadata=hydrate, include_resources=True + ) + + def list_webhook_integrations( + self, + filter_model: WebhookIntegrationFilter, + hydrate: bool = False, + ) -> Page[WebhookIntegrationResponse]: + """List webhook integrations in the active project. + + Args: + filter_model: The webhook integration filters. + hydrate: Whether to include intake statistics. + + Returns: + A page of webhook integrations. + """ + with Session(self.engine) as session: + self._set_filter_project_id( + filter_model=filter_model, session=session + ) + return self.filter_and_paginate( + session=session, + query=select(WebhookIntegrationSchema), + table=WebhookIntegrationSchema, + filter_model=filter_model, + hydrate=hydrate, + apply_query_options_from_schema=True, + ) + + def update_webhook_integration( + self, + integration_id: UUID, + update: WebhookIntegrationUpdate, + ) -> WebhookIntegrationResponse: + """Update a webhook integration. + + Args: + integration_id: The webhook integration ID. + update: The webhook integration update. + + Returns: + The updated webhook integration. + """ + with Session(self.engine) as session: + schema = self._get_schema_by_id( + resource_id=integration_id, + schema_class=WebhookIntegrationSchema, + session=session, + ) + self._verify_name_uniqueness( + resource=update, schema=schema, session=session + ) + schema.update(update) + session.add(schema) + session.commit() + return schema.to_model( + include_metadata=True, include_resources=True + ) + + def delete_webhook_integration(self, integration_id: UUID) -> None: + """Delete a webhook integration and its internal secret. + + Args: + integration_id: The webhook integration ID. + """ + with Session(self.engine) as session: + schema = self._get_schema_by_id( + resource_id=integration_id, + schema_class=WebhookIntegrationSchema, + session=session, + ) + secret_id = schema.secret_id + session.delete(schema) + session.commit() + self._delete_secret_schema(secret_id=secret_id, session=session) + + def rotate_webhook_integration_secret( + self, + integration_id: UUID, + request: WebhookIntegrationRotateSecretRequest, + ) -> WebhookIntegrationSecretResponse: + """Replace the active webhook signing secret. + + Args: + integration_id: The webhook integration ID. + request: The secret rotation request. + + Returns: + The newly active signing secret. + """ + with Session(self.engine) as session: + schema = self._get_schema_by_id( + resource_id=integration_id, + schema_class=WebhookIntegrationSchema, + session=session, + ) + secret = ( + request.secret.get_secret_value() + if request.secret is not None + else secrets.token_urlsafe(32) + ) + self._update_secret_values( + secret_id=schema.secret_id, + values={_WEBHOOK_SECRET_VALUE_KEY: secret}, + overwrite=True, + ) + schema.updated = utc_now() + session.add(schema) + session.commit() + return WebhookIntegrationSecretResponse( + secret=PlainSerializedSecretStr(secret) + ) + + def get_webhook_intake_config( + self, integration_id: UUID + ) -> Tuple[WebhookType, bool, UUID]: + """Get the minimal configuration required for webhook intake. + + Args: + integration_id: The webhook integration ID. + + Returns: + The webhook type, active state, and internal secret ID. + + Raises: + KeyError: If the webhook integration does not exist. + """ + with Session(self.engine) as session: + config = session.exec( + select( + WebhookIntegrationSchema.webhook_type, + WebhookIntegrationSchema.active, + WebhookIntegrationSchema.secret_id, + ).where(col(WebhookIntegrationSchema.id) == integration_id) + ).first() + if config is None: + raise KeyError(f"Webhook integration {integration_id} not found.") + webhook_type, active, secret_id = config + return WebhookType(webhook_type), active, secret_id + + def get_webhook_secret(self, secret_id: UUID) -> str: + """Get a webhook signing secret from the configured secrets store. + + Args: + secret_id: The internal webhook secret ID. + + Returns: + The signing secret. + """ + return self._get_secret_values(secret_id)[_WEBHOOK_SECRET_VALUE_KEY] + + def record_webhook_event( + self, integration_id: UUID, update: WebhookEventStatsUpdate + ) -> None: + """Record a webhook intake outcome. + + Args: + integration_id: The webhook integration ID. + update: The terminal intake outcome. + + Raises: + KeyError: If the webhook integration no longer exists. + """ + now = utc_now() + values: Dict[str, Any] = { + "received_count": ( + WebhookIntegrationStatsSchema.received_count + 1 + ), + "last_received_at": now, + } + if update.accepted: + values["accepted_count"] = ( + WebhookIntegrationStatsSchema.accepted_count + 1 + ) + values["last_accepted_at"] = now + elif update.auth_failed: + values["auth_failed_count"] = ( + WebhookIntegrationStatsSchema.auth_failed_count + 1 + ) + elif update.invalid_payload: + values["invalid_payload_count"] = ( + WebhookIntegrationStatsSchema.invalid_payload_count + 1 + ) + if update.error_summary is not None: + values["last_error_at"] = now + values["last_error_summary"] = update.error_summary + + with Session(self.engine) as session: + result = session.exec( + sqlalchemy.update(WebhookIntegrationStatsSchema) + .where( + col(WebhookIntegrationStatsSchema.webhook_id) + == integration_id + ) + .values(**values) + ) + if result.rowcount == 0: + raise KeyError( + f"Webhook integration {integration_id} not found." + ) + session.commit() + # -------------------- Triggers --------------------- @track_decorator(AnalyticsEvent.CREATED_TRIGGER) @@ -9612,7 +9933,24 @@ def _delete_secret_schema(self, secret_id: UUID, session: Session) -> None: Args: secret_id: The ID of the secret to delete. session: The session to use. - """ + + Raises: + IllegalOperationError: If the secret is owned by a webhook. + """ + # Avoid flushing pending resource deletions before the secrets store, + # which may use a separate connection to the same SQLite database. + with session.no_autoflush: + webhook_id = session.exec( + select(WebhookIntegrationSchema.id).where( + WebhookIntegrationSchema.secret_id == secret_id + ) + ).first() + if webhook_id is not None: + raise IllegalOperationError( + f"Cannot delete secret {secret_id}: it is owned by webhook " + f"{webhook_id}. Delete the webhook instead." + ) + # Delete the secret values in the configured secrets store try: self._delete_secret_values(secret_id=secret_id) @@ -13476,8 +13814,17 @@ def delete_project(self, project_name_or_id: Union[str, UUID]) -> None: "The default project cannot be deleted." ) + webhook_secret_ids = session.exec( + select(WebhookIntegrationSchema.secret_id).where( + WebhookIntegrationSchema.project_id == project.id + ) + ).all() session.delete(project) session.commit() + for secret_id in webhook_secret_ids: + self._delete_secret_schema( + secret_id=secret_id, session=session + ) def count_projects( self, filter_model: Optional[ProjectFilter] = None diff --git a/src/zenml/zen_stores/zen_store_interface.py b/src/zenml/zen_stores/zen_store_interface.py index f15f7467ba5..7daf7bd69d3 100644 --- a/src/zenml/zen_stores/zen_store_interface.py +++ b/src/zenml/zen_stores/zen_store_interface.py @@ -167,6 +167,13 @@ UserRequest, UserResponse, UserUpdate, + WebhookIntegrationCreateResponse, + WebhookIntegrationFilter, + WebhookIntegrationRequest, + WebhookIntegrationResponse, + WebhookIntegrationRotateSecretRequest, + WebhookIntegrationSecretResponse, + WebhookIntegrationUpdate, ) from zenml.zen_stores.resource_pools.store_interface import ( ResourcePoolsStoreInterface, @@ -1824,6 +1831,91 @@ def create_run_metadata(self, run_metadata: RunMetadataRequest) -> None: None """ + # -------------------- Webhook integrations --------------------- + + @abstractmethod + def create_webhook_integration( + self, integration: WebhookIntegrationRequest + ) -> WebhookIntegrationCreateResponse: + """Create a webhook integration. + + Args: + integration: The webhook integration creation request. + + Returns: + The created integration and any generated signing secret. + """ + + @abstractmethod + def get_webhook_integration( + self, integration_id: UUID, hydrate: bool = True + ) -> WebhookIntegrationResponse: + """Get a webhook integration. + + Args: + integration_id: The webhook integration ID. + hydrate: Whether to include intake statistics. + + Returns: + The webhook integration. + """ + + @abstractmethod + def list_webhook_integrations( + self, + filter_model: WebhookIntegrationFilter, + hydrate: bool = False, + ) -> Page[WebhookIntegrationResponse]: + """List webhook integrations. + + Args: + filter_model: The webhook integration filters. + hydrate: Whether to include intake statistics. + + Returns: + A page of webhook integrations. + """ + + @abstractmethod + def update_webhook_integration( + self, + integration_id: UUID, + update: WebhookIntegrationUpdate, + ) -> WebhookIntegrationResponse: + """Update a webhook integration. + + Args: + integration_id: The webhook integration ID. + update: The webhook integration update. + + Returns: + The updated webhook integration. + """ + + @abstractmethod + def delete_webhook_integration(self, integration_id: UUID) -> None: + """Delete a webhook integration and its signing secret. + + Args: + integration_id: The webhook integration ID. + """ + + @abstractmethod + def rotate_webhook_integration_secret( + self, + integration_id: UUID, + request: WebhookIntegrationRotateSecretRequest, + ) -> WebhookIntegrationSecretResponse: + """Rotate a webhook integration signing secret. + + Args: + integration_id: The webhook integration ID. + request: The secret rotation request. + + Returns: + The newly active signing secret. + """ + # -------------------- Triggers --------------------- @abstractmethod diff --git a/tests/integration/functional/webhooks/__init__.py b/tests/integration/functional/webhooks/__init__.py new file mode 100644 index 00000000000..dbee9dcabb7 --- /dev/null +++ b/tests/integration/functional/webhooks/__init__.py @@ -0,0 +1,14 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""Functional tests for webhook integrations.""" diff --git a/tests/integration/functional/webhooks/test_cli.py b/tests/integration/functional/webhooks/test_cli.py new file mode 100644 index 00000000000..e9ba34d3d7b --- /dev/null +++ b/tests/integration/functional/webhooks/test_cli.py @@ -0,0 +1,140 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +import io +from unittest.mock import patch + +import pytest + +import zenml_cli +from tests.cli_runner_utils import cli_runner +from tests.integration.functional.utils import sample_name +from zenml.cli.cli import cli +from zenml.client import Client +from zenml.enums import WebhookType +from zenml.zen_stores.sql_zen_store import SqlZenStore + + +@pytest.fixture +def clean_client(clean_project: Client) -> Client: + """Return the active client when it is connected to a server.""" + if isinstance(clean_project.zen_store, SqlZenStore): + pytest.skip("Webhooks require a REST store.") + return clean_project + + +webhook_command = cli.commands["webhook"] +create_command = webhook_command.commands["create"] +describe_command = webhook_command.commands["describe"] +list_command = webhook_command.commands["list"] +update_command = webhook_command.commands["update"] +rotate_secret_command = webhook_command.commands["rotate-secret"] +delete_command = webhook_command.commands["delete"] + + +def _delete_if_exists(name_or_id: str) -> None: + client = Client() + try: + client.delete_webhook(name_or_id) + except KeyError: + pass + + +def test_webhook_cli_lifecycle(clean_client): + runner = cli_runner() + name = sample_name("webhook-cli") + updated_name = sample_name("webhook-cli-updated") + + try: + result = runner.invoke( + create_command, + [name, "--type", WebhookType.CUSTOM.value], + ) + + assert result.exit_code == 0, result.output + assert "Signing secret:" in result.output + + integration = clean_client.get_webhook(name) + assert integration.webhook_type == WebhookType.CUSTOM + assert integration.active is True + + list_output_buffer = io.StringIO() + with patch.object(zenml_cli, "_original_stdout", list_output_buffer): + result = runner.invoke(list_command) + list_output = list_output_buffer.getvalue() + result.output + + assert result.exit_code == 0, result.output + assert name in list_output + assert "Unknown column(s) ignored" not in list_output + + result = runner.invoke(describe_command, [name]) + + assert result.exit_code == 0, result.output + assert name in result.output + assert integration.endpoint_path in result.output + + result = runner.invoke( + update_command, + [name, "--name", updated_name, "--inactive"], + ) + + assert result.exit_code == 0, result.output + updated = clean_client.get_webhook(updated_name) + assert updated.id == integration.id + assert updated.active is False + + result = runner.invoke( + rotate_secret_command, + [updated_name, "--secret", "replacement-secret"], + ) + + assert result.exit_code == 0, result.output + assert "Signing secret: replacement-secret" in result.output + + result = runner.invoke(delete_command, [updated_name, "--yes"]) + + assert result.exit_code == 0, result.output + with pytest.raises(KeyError): + clean_client.get_webhook(updated.id) + finally: + _delete_if_exists(updated_name) + _delete_if_exists(name) + + +def test_webhook_cli_does_not_echo_user_supplied_secret( + clean_client, +): + runner = cli_runner() + name = sample_name("webhook-cli-secret") + + try: + result = runner.invoke( + create_command, + [ + name, + "--type", + WebhookType.GITHUB.value, + "--secret", + "user-supplied-secret", + ], + ) + + assert result.exit_code == 0, result.output + assert "user-supplied-secret" not in result.output + assert "Signing secret:" not in result.output + + integration = clean_client.get_webhook(name) + assert integration.webhook_type == WebhookType.GITHUB + assert "secret" not in integration.model_dump() + finally: + _delete_if_exists(name) diff --git a/tests/integration/functional/webhooks/test_client.py b/tests/integration/functional/webhooks/test_client.py new file mode 100644 index 00000000000..f0550f36582 --- /dev/null +++ b/tests/integration/functional/webhooks/test_client.py @@ -0,0 +1,138 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +import pytest + +from tests.integration.functional.utils import sample_name +from zenml.client import Client +from zenml.enums import WebhookType +from zenml.zen_stores.sql_zen_store import SqlZenStore + + +@pytest.fixture +def clean_client(clean_project: Client) -> Client: + """Return the active client when it is connected to a server.""" + if isinstance(clean_project.zen_store, SqlZenStore): + pytest.skip("Webhooks require a REST store.") + return clean_project + + +def test_client_webhook_lifecycle(clean_client): + name = sample_name("webhook-client") + + result = clean_client.create_webhook( + name=name, + webhook_type=WebhookType.CUSTOM, + ) + + assert result.secret is not None + integration = result.webhook + assert integration.name == name + assert integration.webhook_type == WebhookType.CUSTOM + assert integration.active is True + assert integration.project_id == clean_client.active_project.id + assert integration.endpoint_path.endswith( + f"/custom/{integration.id}/events" + ) + assert integration.stats.received_count == 0 + + by_id = clean_client.get_webhook(integration.id) + by_name = clean_client.get_webhook(name) + + assert by_id.id == integration.id + assert by_name.id == integration.id + assert "secret" not in by_id.model_dump() + + listed_by_type = clean_client.list_webhooks( + webhook_type=WebhookType.CUSTOM + ) + listed_by_active_state = clean_client.list_webhooks(active=True) + + assert integration.id in {item.id for item in listed_by_type.items} + assert integration.id in {item.id for item in listed_by_active_state.items} + + updated_name = sample_name("webhook-client-updated") + updated = clean_client.update_webhook( + name_id_or_prefix=name, + name=updated_name, + active=False, + ) + + assert updated.id == integration.id + assert updated.name == updated_name + assert updated.active is False + + inactive_integrations = clean_client.list_webhooks(active=False) + + assert integration.id in {item.id for item in inactive_integrations.items} + + rotated = clean_client.rotate_webhook_secret( + name_id_or_prefix=updated_name, + secret="replacement-secret", + ) + + assert rotated.secret.get_secret_value() == "replacement-secret" + + clean_client.delete_webhook(updated_name) + + with pytest.raises(KeyError): + clean_client.get_webhook(integration.id) + + +def test_client_does_not_echo_user_supplied_webhook_secret(clean_client): + name = sample_name("webhook-client-secret") + + result = clean_client.create_webhook( + name=name, + webhook_type=WebhookType.GITHUB, + secret="user-supplied-secret", + ) + + try: + assert result.secret is None + + integration = clean_client.get_webhook(result.webhook.id) + + assert "secret" not in integration.model_dump() + assert integration.webhook_type == WebhookType.GITHUB + finally: + clean_client.delete_webhook(result.webhook.id) + + +def test_client_update_webhook_by_name_and_id( + clean_client, +) -> None: + """Webhook integrations can be updated by name and ID.""" + name = sample_name("webhook-client-update") + result = clean_client.create_webhook( + name=name, + webhook_type=WebhookType.CUSTOM, + ) + integration_id = result.webhook.id + + try: + updated_by_name = clean_client.update_webhook( + name, + active=False, + ) + updated_by_id = clean_client.update_webhook( + integration_id, + active=True, + ) + + assert updated_by_name.id == integration_id + assert updated_by_name.active is False + assert updated_by_id.id == integration_id + assert updated_by_id.active is True + finally: + clean_client.delete_webhook(integration_id) diff --git a/tests/integration/functional/webhooks/test_events_endpoint.py b/tests/integration/functional/webhooks/test_events_endpoint.py new file mode 100644 index 00000000000..8aeb56c644f --- /dev/null +++ b/tests/integration/functional/webhooks/test_events_endpoint.py @@ -0,0 +1,191 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +import hashlib +import hmac +from uuid import uuid4 + +import pytest +import requests + +from tests.integration.functional.utils import sample_name +from zenml.enums import WebhookType +from zenml.models import WebhookIntegrationCreateResponse +from zenml.zen_stores.rest_zen_store import RestZenStore + + +def _require_rest_store(clean_project) -> RestZenStore: + store = clean_project.zen_store + if not isinstance(store, RestZenStore): + pytest.skip("Webhook intake endpoint tests require a REST store.") + return store + + +def _signature(secret: str, body: bytes) -> str: + return ( + "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + ) + + +def _post_webhook( + store: RestZenStore, + endpoint_path: str, + body: bytes, + headers: dict[str, str], +) -> requests.Response: + return requests.post( + store.url + endpoint_path, + data=body, + headers=headers, + timeout=31, + ) + + +@pytest.fixture +def webhook_integration_factory(clean_project): + integration_ids = [] + + def create( + name_prefix: str, + *, + active: bool = True, + secret: str | None = None, + ) -> WebhookIntegrationCreateResponse: + result = clean_project.create_webhook( + name=sample_name(name_prefix), + webhook_type=WebhookType.CUSTOM, + active=active, + secret=secret, + ) + integration_ids.append(result.webhook.id) + return result + + yield create + + for integration_id in integration_ids: + clean_project.delete_webhook(integration_id) + + +def test_webhook_intake_accepts_valid_custom_delivery( + clean_project, webhook_integration_factory +): + store = _require_rest_store(clean_project) + result = webhook_integration_factory("webhook-intake-valid") + integration = result.webhook + assert result.secret is not None + secret = result.secret.get_secret_value() + body = b'{"pipeline":"training"}' + + response = _post_webhook( + store=store, + endpoint_path=integration.endpoint_path, + body=body, + headers={ + "X-ZenML-Event": "pipeline.ready", + "X-ZenML-Delivery": "delivery-1", + "X-ZenML-Signature-256": _signature(secret, body), + }, + ) + + assert response.status_code == 202 + assert response.content == b"" + + updated = clean_project.get_webhook(integration.id) + assert updated.stats.received_count == 1 + assert updated.stats.accepted_count == 1 + assert updated.stats.auth_failed_count == 0 + assert updated.stats.invalid_payload_count == 0 + assert updated.stats.last_received_at is not None + assert updated.stats.last_accepted_at is not None + + +@pytest.mark.parametrize( + ( + "scenario", + "expected_status", + "expected_counts", + "expected_error", + ), + [ + ("auth-failure", 401, (1, 0, 1, 0), "Invalid webhook signature."), + ( + "invalid-payload", + 400, + (1, 0, 0, 1), + "Request body must be valid JSON.", + ), + ("inactive", 409, (0, 0, 0, 0), None), + ("type-mismatch", 404, (0, 0, 0, 0), None), + ], +) +def test_webhook_intake_failure_scenarios( + clean_project, + webhook_integration_factory, + scenario: str, + expected_status: int, + expected_counts: tuple[int, int, int, int], + expected_error: str | None, +): + store = _require_rest_store(clean_project) + result = webhook_integration_factory( + f"webhook-intake-{scenario}", + active=scenario != "inactive", + ) + integration = result.webhook + assert result.secret is not None + secret = result.secret.get_secret_value() + body = b"not-json" if scenario == "invalid-payload" else b'{"ok":true}' + endpoint_path = integration.endpoint_path + headers = { + "X-ZenML-Event": "pipeline.ready", + "X-ZenML-Signature-256": _signature(secret, body), + } + if scenario == "auth-failure": + headers["X-ZenML-Signature-256"] = "sha256=invalid" + elif scenario == "type-mismatch": + endpoint_path = endpoint_path.replace("/custom/", "/github/") + headers = {} + + response = _post_webhook( + store=store, + endpoint_path=endpoint_path, + body=body, + headers=headers, + ) + + assert response.status_code == expected_status + updated = clean_project.get_webhook(integration.id) + assert ( + updated.stats.received_count, + updated.stats.accepted_count, + updated.stats.auth_failed_count, + updated.stats.invalid_payload_count, + ) == expected_counts + assert updated.stats.last_error_summary == expected_error + assert (updated.stats.last_error_at is not None) == ( + expected_error is not None + ) + + +def test_webhook_intake_returns_not_found_for_unknown_integration( + clean_project, +): + store = _require_rest_store(clean_project) + response = _post_webhook( + store=store, + endpoint_path=f"/api/v1/webhooks/custom/{uuid4()}/events", + body=b'{"pipeline":"training"}', + headers={}, + ) + + assert response.status_code == 404 diff --git a/tests/integration/functional/webhooks/test_zen_store.py b/tests/integration/functional/webhooks/test_zen_store.py new file mode 100644 index 00000000000..351526fc23e --- /dev/null +++ b/tests/integration/functional/webhooks/test_zen_store.py @@ -0,0 +1,269 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +import pytest +from sqlalchemy import event +from sqlalchemy.exc import IntegrityError +from sqlmodel import Session, select + +from tests.integration.functional.utils import sample_name +from zenml.enums import WebhookType +from zenml.exceptions import IllegalOperationError +from zenml.models import ( + WebhookEventStatsUpdate, + WebhookIntegrationFilter, + WebhookIntegrationRequest, + WebhookIntegrationRotateSecretRequest, + WebhookIntegrationUpdate, +) +from zenml.zen_stores.schemas.secret_schemas import SecretSchema +from zenml.zen_stores.schemas.webhook_integration_schemas import ( + WebhookIntegrationSchema, +) +from zenml.zen_stores.sql_zen_store import SqlZenStore + + +def test_client_webhook_methods_require_rest_store( + clean_client, +) -> None: + """Public webhook client methods reject local SQL stores.""" + if not isinstance(clean_client.zen_store, SqlZenStore): + pytest.skip("Local SQL store behavior is required for this test.") + + error = "This method is not allowed when not connected" + with pytest.raises(TypeError, match=error): + clean_client.create_webhook( + name="webhook", + webhook_type=WebhookType.CUSTOM, + ) + with pytest.raises(TypeError, match=error): + clean_client.get_webhook("webhook") + with pytest.raises(TypeError, match=error): + clean_client.list_webhooks() + with pytest.raises(TypeError, match=error): + clean_client.update_webhook("webhook", active=False) + with pytest.raises(TypeError, match=error): + clean_client.delete_webhook("webhook") + with pytest.raises(TypeError, match=error): + clean_client.rotate_webhook_secret("webhook") + + +def test_zen_store_webhook_integration_lifecycle(clean_client): + store = clean_client.zen_store + project_id = clean_client.active_project.id + name = sample_name("webhook-store") + + result = store.create_webhook_integration( + WebhookIntegrationRequest( + project=project_id, + name=name, + webhook_type=WebhookType.CUSTOM, + ) + ) + + integration = result.webhook + + assert result.secret is not None + assert integration.name == name + assert integration.project_id == project_id + assert integration.webhook_type == WebhookType.CUSTOM + assert integration.active is True + assert integration.stats.received_count == 0 + assert integration.get_resources().user is not None + assert integration.get_resources().user.id == clean_client.active_user.id + + store.record_webhook_event( + integration.id, WebhookEventStatsUpdate(accepted=True) + ) + integration = store.get_webhook_integration(integration.id) + + assert integration.stats.received_count == 1 + assert integration.stats.accepted_count == 1 + + with Session(store.engine) as session: + initial_secret_id = session.exec( + select(WebhookIntegrationSchema.secret_id).where( + WebhookIntegrationSchema.id == integration.id + ) + ).one() + + by_id = store.get_webhook_integration(integration.id) + + assert by_id.id == integration.id + assert by_id.stats.received_count == 1 + assert by_id.get_resources().user is not None + assert by_id.get_resources().user.id == clean_client.active_user.id + + filtered = store.list_webhook_integrations( + WebhookIntegrationFilter( + project=project_id, + webhook_type=WebhookType.CUSTOM, + active=True, + ), + hydrate=True, + ) + + assert integration.id in {item.id for item in filtered.items} + filtered_integration = next( + item for item in filtered.items if item.id == integration.id + ) + assert filtered_integration.stats.received_count == 1 + + updated_name = sample_name("webhook-store-updated") + updated = store.update_webhook_integration( + integration_id=integration.id, + update=WebhookIntegrationUpdate( + name=updated_name, + active=False, + ), + ) + + assert updated.id == integration.id + assert updated.name == updated_name + assert updated.active is False + assert updated.get_resources().user is not None + assert updated.get_resources().user.id == clean_client.active_user.id + + with Session(store.engine) as session: + updated_secret_id = session.exec( + select(WebhookIntegrationSchema.secret_id).where( + WebhookIntegrationSchema.id == integration.id + ) + ).one() + assert updated_secret_id == initial_secret_id + + inactive_integrations = store.list_webhook_integrations( + WebhookIntegrationFilter(project=project_id, active=False) + ) + + assert integration.id in {item.id for item in inactive_integrations.items} + + rotated = store.rotate_webhook_integration_secret( + integration_id=integration.id, + request=WebhookIntegrationRotateSecretRequest( + secret="replacement-secret" + ), + ) + + assert rotated.secret.get_secret_value() == "replacement-secret" + + with Session(store.engine) as session: + rotated_secret_id = session.exec( + select(WebhookIntegrationSchema.secret_id).where( + WebhookIntegrationSchema.id == integration.id + ) + ).one() + assert rotated_secret_id == initial_secret_id + assert store.get_webhook_secret(rotated_secret_id) == "replacement-secret" + + store.delete_webhook_integration(integration.id) + + with pytest.raises(KeyError): + store.get_webhook_integration(integration.id) + + +def test_sql_store_webhook_intake_uses_one_config_read_and_stats_update( + clean_client, +) -> None: + """Webhook intake uses one SQL statement per database operation.""" + store = clean_client.zen_store + if not isinstance(store, SqlZenStore): + pytest.skip("Local SQL store behavior is required for this test.") + + result = store.create_webhook_integration( + WebhookIntegrationRequest( + project=clean_client.active_project.id, + name=sample_name("webhook-intake-query-count"), + webhook_type=WebhookType.CUSTOM, + ) + ) + statements = [] + + def capture_statement( + connection, cursor, statement, parameters, context, executemany + ): + statements.append(statement) + + event.listen(store.engine, "before_cursor_execute", capture_statement) + try: + webhook_type, active, _ = store.get_webhook_intake_config( + result.webhook.id + ) + + assert webhook_type == WebhookType.CUSTOM + assert active is True + assert len(statements) == 1 + assert statements[0].lstrip().upper().startswith("SELECT") + + statements.clear() + store.record_webhook_event( + result.webhook.id, WebhookEventStatsUpdate(accepted=True) + ) + + assert len(statements) == 1 + assert statements[0].lstrip().upper().startswith("UPDATE") + finally: + event.remove(store.engine, "before_cursor_execute", capture_statement) + store.delete_webhook_integration(result.webhook.id) + + +def test_sql_store_protects_webhook_owned_secret(clean_client) -> None: + """Webhook-owned secrets require explicit webhook deletion.""" + store = clean_client.zen_store + if not isinstance(store, SqlZenStore): + pytest.skip("Local SQL store behavior is required for this test.") + + result = store.create_webhook_integration( + WebhookIntegrationRequest( + project=clean_client.active_project.id, + name=sample_name("webhook-owned-secret"), + webhook_type=WebhookType.CUSTOM, + secret="owned-secret", + ) + ) + webhook_id = result.webhook.id + + try: + with Session(store.engine) as session: + schema = session.exec( + select(WebhookIntegrationSchema).where( + WebhookIntegrationSchema.id == webhook_id + ) + ).one() + secret_id = schema.secret_id + + secret = session.get(SecretSchema, secret_id) + assert secret is not None + session.delete(secret) + with pytest.raises(IntegrityError): + session.commit() + session.rollback() + + with pytest.raises(IllegalOperationError): + store._delete_secret_schema( + secret_id=secret_id, session=session + ) + + assert store.get_webhook_secret(secret_id) == "owned-secret" + assert store.get_webhook_integration(webhook_id).id == webhook_id + + store.delete_webhook_integration(webhook_id) + + with Session(store.engine) as session: + assert session.get(WebhookIntegrationSchema, webhook_id) is None + assert session.get(SecretSchema, secret_id) is None + finally: + try: + store.delete_webhook_integration(webhook_id) + except KeyError: + pass diff --git a/tests/integration/integrations/trackio/experiment_trackers/test_trackio_experiment_tracker.py b/tests/integration/integrations/trackio/experiment_trackers/test_trackio_experiment_tracker.py index 7db7d5d9d7a..80b71837c8a 100644 --- a/tests/integration/integrations/trackio/experiment_trackers/test_trackio_experiment_tracker.py +++ b/tests/integration/integrations/trackio/experiment_trackers/test_trackio_experiment_tracker.py @@ -323,7 +323,6 @@ def test_trackio_cleanup_calls_finish_when_sync_fails( def test_trackio_metadata() -> None: - tracker = _tracker(space_id="AINovice2005/zenml-F5J4jOGLJ44") tracker.get_settings = MagicMock(return_value=_settings()) diff --git a/tests/unit/utils/test_secret_utils.py b/tests/unit/utils/test_secret_utils.py index 3c5e848f0aa..05463ee9a1a 100644 --- a/tests/unit/utils/test_secret_utils.py +++ b/tests/unit/utils/test_secret_utils.py @@ -11,10 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express # or implied. See the License for the specific language governing # permissions and limitations under the License. +"""Unit tests for secret utilities.""" +import pytest from hypothesis import given from hypothesis.strategies import from_regex -from pydantic import BaseModel, Field, SecretStr +from pydantic import BaseModel, Field, SecretStr, ValidationError from zenml.utils import secret_utils @@ -90,3 +92,30 @@ class Model(BaseModel): secret_utils.is_secret_field(Model.model_fields["non_secret"]) is False ) assert secret_utils.is_secret_field(Model.model_fields["secret"]) is True + + +@pytest.mark.parametrize("value", ["", " ", "\t\n"]) +def test_non_empty_plain_serialized_secret_str_rejects_empty_values( + value: str, +) -> None: + """Tests that non-empty serialized secrets reject blank values.""" + + class Model(BaseModel): + secret: secret_utils.NonEmptyPlainSerializedSecretStr + + with pytest.raises(ValidationError): + Model(secret=value) + + +def test_non_empty_plain_serialized_secret_str_serializes_plain_value() -> ( + None +): + """Tests that non-empty serialized secrets preserve plain JSON output.""" + + class Model(BaseModel): + secret: secret_utils.NonEmptyPlainSerializedSecretStr + + model = Model(secret=" value ") + + assert model.secret.get_secret_value() == " value " + assert model.model_dump_json() == '{"secret":" value "}' diff --git a/tests/unit/webhooks/test_adapters.py b/tests/unit/webhooks/test_adapters.py new file mode 100644 index 00000000000..e733a7b7993 --- /dev/null +++ b/tests/unit/webhooks/test_adapters.py @@ -0,0 +1,190 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +import hashlib +import hmac +from collections.abc import Mapping + +import pytest + +from zenml.enums import WebhookType +from zenml.webhooks.adapters import ( + BaseWebhookAdapter, + CustomWebhookAdapter, + GitHubWebhookAdapter, + WebhookAuthenticationError, + WebhookPayloadError, + get_webhook_adapter, +) + + +class _BodyMetadataBearerAdapter(BaseWebhookAdapter): + """Test adapter with bearer auth and body-derived event metadata.""" + + webhook_type = WebhookType.CUSTOM + + def authenticate( + self, body: bytes, headers: Mapping[str, str], secret: str + ) -> None: + if headers.get("authorization") != f"Bearer {secret}": + raise WebhookAuthenticationError("Invalid bearer token.") + + def get_event_type(self, payload: dict, headers: Mapping[str, str]) -> str: + event_type = payload.get("type") + if not isinstance(event_type, str) or not event_type: + raise WebhookPayloadError("Missing event type in request body.") + return event_type + + def get_delivery_id( + self, payload: dict, headers: Mapping[str, str] + ) -> str | None: + delivery_id = payload.get("webhookId") + return delivery_id if isinstance(delivery_id, str) else None + + +def _signature(secret: str, body: bytes) -> str: + return ( + "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + ) + + +@pytest.mark.parametrize( + "webhook_type, adapter_type", + [ + (WebhookType.GITHUB, GitHubWebhookAdapter), + (WebhookType.CUSTOM, CustomWebhookAdapter), + ], +) +def test_get_webhook_adapter_returns_registered_adapter( + webhook_type: WebhookType, adapter_type: type +) -> None: + adapter = get_webhook_adapter(webhook_type) + + assert isinstance(adapter, adapter_type) + assert adapter.webhook_type == webhook_type + + +@pytest.mark.parametrize( + "adapter, headers, expected_event_type, expected_delivery_id", + [ + ( + GitHubWebhookAdapter(), + { + "x-github-event": "push", + "x-github-delivery": "github-delivery-id", + }, + "push", + "github-delivery-id", + ), + ( + CustomWebhookAdapter(), + { + "x-zenml-event": "pipeline.ready", + "x-zenml-delivery": "custom-delivery-id", + }, + "pipeline.ready", + "custom-delivery-id", + ), + ], +) +def test_validate_returns_provider_event_for_signed_request( + adapter: GitHubWebhookAdapter | CustomWebhookAdapter, + headers: dict[str, str], + expected_event_type: str, + expected_delivery_id: str, +) -> None: + secret = "webhook-secret" + body = b'{"repository":"zenml","run_id":42}' + headers[adapter.signature_header] = _signature(secret, body) + + event = adapter.validate(body=body, headers=headers, secret=secret) + + assert event.webhook_type == adapter.webhook_type + assert event.event_type == expected_event_type + assert event.delivery_id == expected_delivery_id + assert event.payload == {"repository": "zenml", "run_id": 42} + + +def test_validate_supports_bearer_auth_and_body_event_metadata() -> None: + adapter = _BodyMetadataBearerAdapter() + body = b'{"type":"monitor.page","webhookId":"delivery-id"}' + + event = adapter.validate( + body=body, + headers={"authorization": "Bearer webhook-secret"}, + secret="webhook-secret", + ) + + assert event.event_type == "monitor.page" + assert event.delivery_id == "delivery-id" + + +def test_parse_rejects_missing_body_event_type() -> None: + adapter = _BodyMetadataBearerAdapter() + + with pytest.raises( + WebhookPayloadError, match="Missing event type in request body" + ): + adapter.parse(body=b'{"webhookId":"delivery-id"}', headers={}) + + +def test_authentication_uses_exact_raw_body_bytes() -> None: + adapter = CustomWebhookAdapter() + secret = "webhook-secret" + signed_body = b'{"repository":"zenml","run_id":42}' + equivalent_body = b'{\n "repository": "zenml",\n "run_id": 42\n}' + headers = { + "x-zenml-event": "pipeline.ready", + "x-zenml-signature-256": _signature(secret, signed_body), + } + + with pytest.raises(WebhookAuthenticationError): + adapter.validate(body=equivalent_body, headers=headers, secret=secret) + + +@pytest.mark.parametrize( + "headers", + [ + {}, + {"x-zenml-signature-256": "not-prefixed"}, + {"x-zenml-signature-256": "sha256=invalid"}, + ], +) +def test_authentication_rejects_missing_malformed_or_invalid_signature( + headers: Mapping[str, str], +) -> None: + adapter = CustomWebhookAdapter() + + with pytest.raises(WebhookAuthenticationError): + adapter.authenticate( + body=b'{"repository":"zenml"}', + headers=headers, + secret="webhook-secret", + ) + + +@pytest.mark.parametrize( + "body, headers", + [ + (b'{"repository":"zenml"}', {}), + (b"not-json", {"x-zenml-event": "pipeline.ready"}), + (b'["not", "an", "object"]', {"x-zenml-event": "pipeline.ready"}), + ], +) +def test_parse_rejects_missing_event_header_or_invalid_payload( + body: bytes, headers: Mapping[str, str] +) -> None: + adapter = CustomWebhookAdapter() + + with pytest.raises(WebhookPayloadError): + adapter.parse(body=body, headers=headers) diff --git a/tests/unit/zen_server/endpoints/test_webhook_integration_endpoints.py b/tests/unit/zen_server/endpoints/test_webhook_integration_endpoints.py new file mode 100644 index 00000000000..fdf8d1ed8f1 --- /dev/null +++ b/tests/unit/zen_server/endpoints/test_webhook_integration_endpoints.py @@ -0,0 +1,194 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from fastapi import HTTPException, status + +from zenml.constants import API, VERSION_1, WEBHOOKS +from zenml.enums import WebhookType +from zenml.webhooks import WebhookAuthenticationError, WebhookPayloadError +from zenml.zen_server.routers import webhook_integration_endpoints as endpoints + + +def test_webhook_routers_use_public_webhook_prefix() -> None: + """Management and intake endpoints share the public webhook prefix.""" + expected_prefix = API + VERSION_1 + WEBHOOKS + + assert endpoints.management_router.prefix == expected_prefix + assert endpoints.intake_router.prefix == expected_prefix + + +class _Store: + def __init__(self, integration: SimpleNamespace | None) -> None: + self.integration = integration + self.secret_requests = 0 + self.records = [] + self.secret_id = uuid4() + + def get_webhook_intake_config(self, integration_id): + if self.integration is None: + raise KeyError(integration_id) + return ( + self.integration.webhook_type, + self.integration.active, + self.secret_id, + ) + + def get_webhook_secret(self, secret_id): + assert secret_id == self.secret_id + self.secret_requests += 1 + return "webhook-secret" + + def record_webhook_event(self, integration_id, update): + self.records.append((integration_id, update)) + + +class _Adapter: + def __init__( + self, + auth_error: Exception | None = None, + payload_error: Exception | None = None, + ) -> None: + self.auth_error = auth_error + self.payload_error = payload_error + self.authenticate_calls = 0 + self.parse_calls = 0 + + def authenticate(self, body, headers, secret): + self.authenticate_calls += 1 + if self.auth_error: + raise self.auth_error + + def parse(self, body, headers): + self.parse_calls += 1 + if self.payload_error: + raise self.payload_error + + +def _install_dependencies(monkeypatch, store: _Store, adapter: _Adapter): + monkeypatch.setattr(endpoints, "zen_store", lambda: store) + monkeypatch.setattr(endpoints, "get_webhook_adapter", lambda _: adapter) + + +def _receive(integration_id): + return endpoints._receive_webhook_event( + webhook_type=WebhookType.CUSTOM, + integration_id=integration_id, + body=b'{"event":"ready"}', + headers={}, + ) + + +@pytest.mark.parametrize( + ( + "stored_type", + "active", + "auth_error", + "payload_error", + "expected_status", + "expected_outcome", + "expected_error", + ), + [ + (None, None, None, None, 404, None, None), + (WebhookType.GITHUB, True, None, None, 404, None, None), + ( + WebhookType.CUSTOM, + True, + WebhookAuthenticationError("bad auth"), + None, + 401, + "auth_failed", + "bad auth", + ), + ( + WebhookType.CUSTOM, + False, + WebhookAuthenticationError("bad auth"), + None, + 401, + None, + None, + ), + (WebhookType.CUSTOM, False, None, None, 409, None, None), + ( + WebhookType.CUSTOM, + True, + None, + WebhookPayloadError("bad payload"), + 400, + "invalid_payload", + "bad payload", + ), + (WebhookType.CUSTOM, True, None, None, 202, "accepted", None), + ], + ids=[ + "missing-integration", + "provider-mismatch", + "active-auth-failure", + "inactive-auth-failure", + "inactive-authenticated", + "invalid-payload", + "accepted", + ], +) +def test_receive_webhook_event_decision_table( + monkeypatch, + stored_type: WebhookType | None, + active: bool | None, + auth_error: Exception | None, + payload_error: Exception | None, + expected_status: int, + expected_outcome: str | None, + expected_error: str | None, +) -> None: + integration_id = uuid4() + store = _Store( + integration=( + SimpleNamespace(webhook_type=stored_type, active=active) + if stored_type is not None + else None + ) + ) + adapter = _Adapter(auth_error=auth_error, payload_error=payload_error) + _install_dependencies(monkeypatch, store, adapter) + + if expected_status == status.HTTP_202_ACCEPTED: + assert _receive(integration_id).status_code == expected_status + else: + with pytest.raises(HTTPException) as error: + _receive(integration_id) + assert error.value.status_code == expected_status + if auth_error is not None: + assert error.value.detail == "Invalid webhook authentication." + + resolved = stored_type == WebhookType.CUSTOM + parsed = expected_status in { + status.HTTP_202_ACCEPTED, + status.HTTP_400_BAD_REQUEST, + } + assert store.secret_requests == int(resolved) + assert adapter.authenticate_calls == int(resolved) + assert adapter.parse_calls == int(parsed) + + if expected_outcome is None: + assert store.records == [] + else: + assert len(store.records) == 1 + recorded_id, update = store.records[0] + assert recorded_id == integration_id + assert getattr(update, expected_outcome) is True + assert update.error_summary == expected_error diff --git a/tests/unit/zen_stores/schemas/test_webhook_integration_schemas.py b/tests/unit/zen_stores/schemas/test_webhook_integration_schemas.py new file mode 100644 index 00000000000..10f0923b4e1 --- /dev/null +++ b/tests/unit/zen_stores/schemas/test_webhook_integration_schemas.py @@ -0,0 +1,174 @@ +# Copyright (c) ZenML GmbH 2026. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing +# permissions and limitations under the License. +"""Unit tests for webhook integration schemas and request models.""" + +from datetime import datetime +from uuid import uuid4 + +import pytest +from pydantic import BaseModel, ValidationError + +from zenml.constants import API, VERSION_1, WEBHOOKS +from zenml.enums import WebhookType +from zenml.models import ( + WebhookEventStatsUpdate, + WebhookIntegrationRequest, + WebhookIntegrationRotateSecretRequest, + WebhookIntegrationStats, + WebhookIntegrationUpdate, +) +from zenml.zen_stores.schemas.webhook_integration_schemas import ( + WebhookIntegrationSchema, + WebhookIntegrationStatsSchema, +) + + +def _webhook_integration_schema() -> WebhookIntegrationSchema: + schema = WebhookIntegrationSchema( + id=uuid4(), + name="github-intake", + project_id=uuid4(), + user_id=uuid4(), + secret_id=uuid4(), + webhook_type=WebhookType.GITHUB.value, + active=True, + ) + schema.stats = WebhookIntegrationStatsSchema( + webhook_id=schema.id, + **WebhookIntegrationStats( + received_count=3, + accepted_count=1, + auth_failed_count=1, + invalid_payload_count=1, + last_received_at=datetime(2026, 7, 9, 8, 0, 0), + last_accepted_at=datetime(2026, 7, 9, 8, 1, 0), + last_error_at=datetime(2026, 7, 9, 8, 2, 0), + last_error_summary="Invalid webhook signature.", + ).model_dump(), + ) + return schema + + +@pytest.mark.parametrize( + "kwargs", + [ + {}, + {"accepted": True, "auth_failed": True}, + {"accepted": True, "error_summary": "unexpected"}, + ], +) +def test_webhook_event_stats_update_rejects_invalid_outcome( + kwargs: dict[str, object], +) -> None: + """Webhook stats updates reject missing or conflicting outcomes.""" + with pytest.raises(ValidationError): + WebhookEventStatsUpdate(**kwargs) + + +@pytest.mark.parametrize( + ("model_class", "kwargs"), + [ + ( + WebhookIntegrationRequest, + { + "name": "github-intake", + "project": uuid4(), + "webhook_type": WebhookType.GITHUB, + }, + ), + (WebhookIntegrationRotateSecretRequest, {}), + ], + ids=["create", "rotate"], +) +@pytest.mark.parametrize("secret", ["", " "]) +def test_webhook_integration_models_reject_empty_secret( + model_class: type[BaseModel], + kwargs: dict[str, object], + secret: str, +) -> None: + """Webhook integration models reject empty signing secrets.""" + with pytest.raises(ValidationError): + model_class(secret=secret, **kwargs) + + +def test_webhook_integration_requests_allow_missing_secret() -> None: + """Webhook integration requests allow missing secrets for generation.""" + integration_request = WebhookIntegrationRequest( + name="github-intake", + project=uuid4(), + webhook_type=WebhookType.GITHUB, + ) + secret_request = WebhookIntegrationRotateSecretRequest() + + assert integration_request.secret is None + assert secret_request.secret is None + + +def test_webhook_integration_update_excludes_secret() -> None: + """Webhook integration updates are limited to database fields.""" + assert "secret" not in WebhookIntegrationUpdate.model_fields + + +def test_webhook_integration_schema_to_model_includes_body_and_metadata() -> ( + None +): + """Webhook integration schemas include body and stats metadata.""" + schema = _webhook_integration_schema() + + response = schema.to_model(include_metadata=True) + + assert response.id == schema.id + assert response.name == "github-intake" + assert response.webhook_type == WebhookType.GITHUB + assert response.active is True + assert response.endpoint_path == ( + f"{API}{VERSION_1}{WEBHOOKS}/{WebhookType.GITHUB.value}/" + f"{schema.id}/events" + ) + assert response.stats.received_count == 3 + assert response.stats.accepted_count == 1 + assert response.stats.auth_failed_count == 1 + assert response.stats.invalid_payload_count == 1 + assert response.stats.last_received_at == datetime(2026, 7, 9, 8, 0, 0) + assert response.stats.last_accepted_at == datetime(2026, 7, 9, 8, 1, 0) + assert response.stats.last_error_at == datetime(2026, 7, 9, 8, 2, 0) + assert response.stats.last_error_summary == "Invalid webhook signature." + + +def test_webhook_integration_stats_schema_defaults_missing_stats() -> None: + """Webhook integration stats schemas default missing fields.""" + schema = _webhook_integration_schema() + schema.stats = WebhookIntegrationStatsSchema( + webhook_id=schema.id, received_count=3 + ) + + response = schema.to_model(include_metadata=True) + + assert response.stats.received_count == 3 + assert response.stats.accepted_count == 0 + assert response.stats.auth_failed_count == 0 + assert response.stats.invalid_payload_count == 0 + assert response.stats.last_received_at is None + + +def test_webhook_integration_schema_to_model_can_include_empty_resources() -> ( + None +): + """Webhook integration schemas can include empty resources.""" + schema = _webhook_integration_schema() + schema.user = None + + response = schema.to_model(include_resources=True) + + assert response.get_resources().user is None