From 709c92c74662d2d852cc3560bbfd095879780a61 Mon Sep 17 00:00:00 2001 From: "Garen J. Torikian" Date: Tue, 10 Nov 2026 09:57:16 -0800 Subject: [PATCH 1/2] feat(actions): Add verify_action returning a typed action context AuthKit Actions requests were only reachable through construct_action, which returns the verified payload as an untyped dict. Customers had to discover field names from example requests and index into the dict by hand, with no type checking or editor support. verify_action verifies the signature the same way and deserializes the payload into AuthenticationActionContext or UserRegistrationActionContext, discriminated on the object field. construct_action, verify_header, and sign_response keep their current signatures and behavior. The Action models are dedicated dataclasses rather than the generated REST models. The API sends a trimmed organization_membership (no roles) and a trimmed invitation (no token or accept_invitation_url), and the dashboard test request strips user and organization further. The generated models' strict from_dict would raise on every real request, so these models require only what the actions-types contract guarantees and treat the production extras as optional. AsyncActions is now a subclass of Actions instead of a duplicated copy. --- README.md | 47 +++- src/workos/actions.py | 612 ++++++++++++++++++++++++++++++++++++------ tests/test_actions.py | 526 ++++++++++++++++++++++++++++++------ 3 files changed, 1029 insertions(+), 156 deletions(-) diff --git a/README.md b/README.md index 1968c29c..c3213289 100644 --- a/README.md +++ b/README.md @@ -88,7 +88,7 @@ The client exposes the WorkOS API through typed namespace properties: | `client.radar` | Radar risk scoring | | `client.passwordless` | Passwordless authentication sessions | | `client.vault` | Encrypted data vault | -| `client.actions` | AuthKit Actions signature verification and response signing | +| `client.actions` | AuthKit Actions request verification, typed deserialization, and response signing | | `client.pkce` | PKCE code verifier/challenge helpers | ## Webhook Signature Verification @@ -121,6 +121,51 @@ def workos_webhook(request): The `WorkOS-Signature` header is formatted `t=, v1=` — note the `, ` separator — and `v1` is the HMAC-SHA256 of `"{timestamp}.{raw body}"` keyed with the endpoint secret. Events older (or newer) than `tolerance` seconds are rejected; it defaults to 180. Verification is local and synchronous, so it works the same on `AsyncWorkOSClient`. +## AuthKit Actions + +[Actions](https://workos.com/docs/authkit/actions) call your endpoint during sign-in or sign-up and wait for a signed verdict. `client.actions.verify_action()` checks the `WorkOS-Signature` header and returns a typed context, and `client.actions.sign_response()` builds the signed reply: + +```python +from django.http import HttpResponse, JsonResponse + +from workos import WorkOSClient +from workos.actions import ActionType, AuthenticationActionContext + +client = WorkOSClient(api_key="sk_1234", client_id="client_1234") +ACTIONS_SECRET = "actions_secret_1234" # from the Actions page in the WorkOS dashboard + +def workos_action(request): + try: + action = client.actions.verify_action( + payload=request.body, # raw bytes, never a re-serialized dict + sig_header=request.headers["WorkOS-Signature"], + secret=ACTIONS_SECRET, + ) + except ValueError: + return HttpResponse(status=400) # invalid signature or stale timestamp + + action_type: ActionType + if isinstance(action, AuthenticationActionContext): + action_type = "authentication" + allowed = action.user.email_verified + else: # UserRegistrationActionContext + action_type = "user_registration" + allowed = action.user_data.email.endswith("@example.com") + + return JsonResponse( + client.actions.sign_response( + action_type=action_type, + verdict="Allow" if allowed else "Deny", + error_message=None if allowed else "Not allowed", + secret=ACTIONS_SECRET, + ) + ) +``` + +`AuthenticationActionContext` carries the `user`, and the `organization` and `organization_membership` when the sign-in targets one. `UserRegistrationActionContext` carries the submitted `user_data` and the `invitation` being accepted, if any. Both have the `id` of the action execution, the `authentication_method`, and the request's `ip_address`, `user_agent`, `device_fingerprint` and `client_id` where WorkOS has them. + +`construct_action()` does the same verification but returns the raw payload as a `dict`, and `verify_header()` only checks the signature. Actions requests older (or newer) than `tolerance` seconds are rejected; it defaults to 30. + ## Pagination Paginated endpoints return `SyncPage[T]` (or `AsyncPage[T]`) with built-in auto-pagination: diff --git a/src/workos/actions.py b/src/workos/actions.py index 258d7986..97c19875 100644 --- a/src/workos/actions.py +++ b/src/workos/actions.py @@ -1,7 +1,17 @@ # @oagen-ignore-file # This file is hand-maintained. AuthKit Actions helpers for request -# verification and response signing. These are client-side cryptographic -# helpers that will always be hand-maintained. +# verification, typed deserialization, and response signing. Actions requests +# are sent by WorkOS to the customer's endpoint and are not described by the +# OpenAPI spec, so both the crypto and the payload models live here. +# +# The payload models mirror the `@workos-inc/actions-types` contract in the +# WorkOS API. Fields that contract guarantees are required; fields WorkOS adds +# on a real sign-in (for example `user.external_id`) are optional, because a +# test request sent from the dashboard omits them. The generated REST models +# (`User`, `Organization`, `OrganizationMembership`, `Invitation`) are not +# reused on purpose: an Action payload carries a trimmed organization +# membership (no `roles`) and a trimmed invitation (no `token`), and their +# strict `from_dict` would reject every real request. from __future__ import annotations @@ -9,7 +19,39 @@ import hmac import json import time -from typing import Any, Dict, Literal, Optional, Union +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Dict, Literal, Optional, Union, cast + +from ._types import _parse_datetime, _raise_deserialize_error +from .common.models.authenticate_response_authentication_method import ( + AuthenticateResponseAuthenticationMethod, +) +from .common.models.invitation_state import InvitationState +from .common.models.organization_domain_state import OrganizationDomainState +from .common.models.organization_domain_verification_strategy import ( + OrganizationDomainVerificationStrategy, +) +from .common.models.organization_membership_status import ( + OrganizationMembershipStatus, +) +from .common.models.slim_role import SlimRole + +__all__ = [ + "DEFAULT_TOLERANCE", + "ActionType", + "ActionContext", + "ActionInvitation", + "ActionOrganization", + "ActionOrganizationDomain", + "ActionOrganizationMembership", + "ActionUser", + "Actions", + "AsyncActions", + "AuthenticationActionContext", + "UserData", + "UserRegistrationActionContext", +] DEFAULT_TOLERANCE = 30 # seconds (stricter than webhooks' 180s) @@ -21,6 +63,377 @@ } +# --------------------------------------------------------------------------- +# Payload models +# --------------------------------------------------------------------------- + + +def _optional_datetime(value: Any) -> datetime | None: + return _parse_datetime(value) if value is not None else None + + +@dataclass(slots=True) +class ActionUser: + """The user signing in, as sent in an Authentication Action request. + + Mirrors the REST ``User`` object. Fields WorkOS adds on a real sign-in + (``name``, ``external_id``, ``last_sign_in_at``, ``locale``, ``metadata``) + are ``None`` when the request omits them, as a dashboard test request does. + """ + + object: Literal["user"] + id: str + email: str + email_verified: bool + first_name: str | None + last_name: str | None + profile_picture_url: str | None + created_at: datetime + updated_at: datetime + name: str | None = None + external_id: str | None = None + last_sign_in_at: datetime | None = None + locale: str | None = None + metadata: dict[str, str] | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ActionUser: + """Deserialize from a dictionary.""" + try: + return cls( + object=data.get("object", "user"), + id=data["id"], + email=data["email"], + email_verified=data["email_verified"], + first_name=data["first_name"], + last_name=data["last_name"], + profile_picture_url=data["profile_picture_url"], + created_at=_parse_datetime(data["created_at"]), + updated_at=_parse_datetime(data["updated_at"]), + name=data.get("name"), + external_id=data.get("external_id"), + last_sign_in_at=_optional_datetime(data.get("last_sign_in_at")), + locale=data.get("locale"), + metadata=data.get("metadata"), + ) + except (KeyError, TypeError, ValueError) as e: + _raise_deserialize_error("ActionUser", e) + + +@dataclass(slots=True) +class ActionOrganizationDomain: + """A domain of the organization in an Authentication Action request.""" + + object: Literal["organization_domain"] + id: str + organization_id: str + domain: str + state: OrganizationDomainState | None = None + verification_strategy: OrganizationDomainVerificationStrategy | None = None + verification_token: str | None = None + verification_prefix: str | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ActionOrganizationDomain: + """Deserialize from a dictionary.""" + try: + state = data.get("state") + strategy = data.get("verification_strategy") + return cls( + object=data.get("object", "organization_domain"), + id=data["id"], + organization_id=data["organization_id"], + domain=data["domain"], + state=OrganizationDomainState(state) if state is not None else None, + verification_strategy=OrganizationDomainVerificationStrategy(strategy) + if strategy is not None + else None, + verification_token=data.get("verification_token"), + verification_prefix=data.get("verification_prefix"), + created_at=_optional_datetime(data.get("created_at")), + updated_at=_optional_datetime(data.get("updated_at")), + ) + except (KeyError, TypeError, ValueError) as e: + _raise_deserialize_error("ActionOrganizationDomain", e) + + +@dataclass(slots=True) +class ActionOrganization: + """The organization the user is signing in to, when there is one. + + Mirrors the REST ``Organization`` object. ``metadata`` and ``external_id`` + are present on a real sign-in and ``None`` on a dashboard test request. + """ + + object: Literal["organization"] + id: str + name: str + domains: list[ActionOrganizationDomain] + created_at: datetime + updated_at: datetime + allow_profiles_outside_organization: bool | None = None + stripe_customer_id: str | None = None + metadata: dict[str, str] | None = None + external_id: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ActionOrganization: + """Deserialize from a dictionary.""" + try: + return cls( + object=data.get("object", "organization"), + id=data["id"], + name=data["name"], + domains=[ + ActionOrganizationDomain.from_dict(cast(dict[str, Any], item)) + for item in cast(list[Any], data["domains"]) + ], + created_at=_parse_datetime(data["created_at"]), + updated_at=_parse_datetime(data["updated_at"]), + allow_profiles_outside_organization=data.get( + "allow_profiles_outside_organization" + ), + stripe_customer_id=data.get("stripe_customer_id"), + metadata=data.get("metadata"), + external_id=data.get("external_id"), + ) + except (KeyError, TypeError, ValueError) as e: + _raise_deserialize_error("ActionOrganization", e) + + +@dataclass(slots=True) +class ActionOrganizationMembership: + """The user's membership in the organization, when there is one. + + This is a trimmed form of the REST ``OrganizationMembership`` object: it + carries the primary ``role`` but not the ``roles`` list. ``directory_managed``, + ``custom_attributes`` and ``user`` are present on a real sign-in and ``None`` + on a dashboard test request. + """ + + object: Literal["organization_membership"] + id: str + user_id: str + organization_id: str + role: SlimRole + status: OrganizationMembershipStatus + created_at: datetime + updated_at: datetime + directory_managed: bool | None = None + custom_attributes: dict[str, Any] | None = None + user: ActionUser | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ActionOrganizationMembership: + """Deserialize from a dictionary.""" + try: + user = data.get("user") + return cls( + object=data.get("object", "organization_membership"), + id=data["id"], + user_id=data["user_id"], + organization_id=data["organization_id"], + role=SlimRole.from_dict(cast(dict[str, Any], data["role"])), + status=OrganizationMembershipStatus(data["status"]), + created_at=_parse_datetime(data["created_at"]), + updated_at=_parse_datetime(data["updated_at"]), + directory_managed=data.get("directory_managed"), + custom_attributes=data.get("custom_attributes"), + user=ActionUser.from_dict(cast(dict[str, Any], user)) + if user is not None + else None, + ) + except (KeyError, TypeError, ValueError) as e: + _raise_deserialize_error("ActionOrganizationMembership", e) + + +@dataclass(slots=True) +class UserData: + """The details a prospective user submitted at sign-up. + + Sent in a User Registration Action request; no user exists yet. + """ + + object: Literal["user_data"] + email: str + first_name: str | None + last_name: str | None + extra_attributes: dict[str, str] | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> UserData: + """Deserialize from a dictionary.""" + try: + return cls( + object=data.get("object", "user_data"), + email=data["email"], + first_name=data["first_name"], + last_name=data["last_name"], + extra_attributes=data.get("extra_attributes"), + ) + except (KeyError, TypeError, ValueError) as e: + _raise_deserialize_error("UserData", e) + + +@dataclass(slots=True) +class ActionInvitation: + """The invitation the prospective user is accepting, when there is one. + + This is a trimmed form of the REST ``Invitation`` object: it has no + ``token``, ``accept_invitation_url``, ``accepted_user_id`` or ``role_slug``. + """ + + object: Literal["invitation"] + id: str + email: str + state: InvitationState + expires_at: datetime + created_at: datetime + updated_at: datetime + accepted_at: datetime | None = None + revoked_at: datetime | None = None + organization_id: str | None = None + inviter_user_id: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ActionInvitation: + """Deserialize from a dictionary.""" + try: + return cls( + object=data.get("object", "invitation"), + id=data["id"], + email=data["email"], + state=InvitationState(data["state"]), + expires_at=_parse_datetime(data["expires_at"]), + created_at=_parse_datetime(data["created_at"]), + updated_at=_parse_datetime(data["updated_at"]), + accepted_at=_optional_datetime(data.get("accepted_at")), + revoked_at=_optional_datetime(data.get("revoked_at")), + organization_id=data.get("organization_id"), + inviter_user_id=data.get("inviter_user_id"), + ) + except (KeyError, TypeError, ValueError) as e: + _raise_deserialize_error("ActionInvitation", e) + + +@dataclass(slots=True) +class AuthenticationActionContext: + """An Authentication Action request: a user is about to sign in. + + ``id`` is the action execution ID, which also appears in the + ``action.authentication_denied`` event and the dashboard's Actions log. + """ + + object: Literal["authentication_action_context"] + id: str + user: ActionUser + authentication_method: AuthenticateResponseAuthenticationMethod + organization: ActionOrganization | None = None + organization_membership: ActionOrganizationMembership | None = None + issuer: str | None = None + ip_address: str | None = None + user_agent: str | None = None + device_fingerprint: str | None = None + client_id: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> AuthenticationActionContext: + """Deserialize from a dictionary.""" + try: + organization = data.get("organization") + membership = data.get("organization_membership") + return cls( + object=data.get("object", "authentication_action_context"), + id=data["id"], + user=ActionUser.from_dict(cast(dict[str, Any], data["user"])), + authentication_method=AuthenticateResponseAuthenticationMethod( + data["authentication_method"] + ), + organization=ActionOrganization.from_dict( + cast(dict[str, Any], organization) + ) + if organization is not None + else None, + organization_membership=ActionOrganizationMembership.from_dict( + cast(dict[str, Any], membership) + ) + if membership is not None + else None, + issuer=data.get("issuer"), + ip_address=data.get("ip_address"), + user_agent=data.get("user_agent"), + device_fingerprint=data.get("device_fingerprint"), + client_id=data.get("client_id"), + ) + except (KeyError, TypeError, ValueError) as e: + _raise_deserialize_error("AuthenticationActionContext", e) + + +@dataclass(slots=True) +class UserRegistrationActionContext: + """A User Registration Action request: someone is about to sign up. + + ``id`` is the action execution ID, which also appears in the + ``action.user_registration_denied`` event and the dashboard's Actions log. + """ + + object: Literal["user_registration_action_context"] + id: str + user_data: UserData + authentication_method: AuthenticateResponseAuthenticationMethod + invitation: ActionInvitation | None = None + ip_address: str | None = None + user_agent: str | None = None + device_fingerprint: str | None = None + client_id: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> UserRegistrationActionContext: + """Deserialize from a dictionary.""" + try: + invitation = data.get("invitation") + return cls( + object=data.get("object", "user_registration_action_context"), + id=data["id"], + user_data=UserData.from_dict(cast(dict[str, Any], data["user_data"])), + authentication_method=AuthenticateResponseAuthenticationMethod( + data["authentication_method"] + ), + invitation=ActionInvitation.from_dict(cast(dict[str, Any], invitation)) + if invitation is not None + else None, + ip_address=data.get("ip_address"), + user_agent=data.get("user_agent"), + device_fingerprint=data.get("device_fingerprint"), + client_id=data.get("client_id"), + ) + except (KeyError, TypeError, ValueError) as e: + _raise_deserialize_error("UserRegistrationActionContext", e) + + +ActionContext = Union[AuthenticationActionContext, UserRegistrationActionContext] +"""A verified Actions request, narrowed by ``isinstance`` on its ``object``.""" + + +def _deserialize_action_context(data: dict[str, Any]) -> ActionContext: + object_type = data.get("object") + if object_type == "authentication_action_context": + return AuthenticationActionContext.from_dict(data) + if object_type == "user_registration_action_context": + return UserRegistrationActionContext.from_dict(data) + _raise_deserialize_error( + "ActionContext", + ValueError(f"Unknown action context object {object_type!r}"), + ) + + +# --------------------------------------------------------------------------- +# Signature helpers +# --------------------------------------------------------------------------- + + def _verify_signature( *, payload: Union[bytes, str], @@ -70,8 +483,66 @@ def _compute_signature(payload_str: str, secret: str) -> str: ).hexdigest() +def _load_verified_payload( + *, + payload: Union[bytes, str], + sig_header: str, + secret: str, + tolerance: int, +) -> Dict[str, Any]: + """Verify the signature, then parse the body as a JSON object.""" + _verify_signature( + payload=payload, + sig_header=sig_header, + secret=secret, + tolerance=tolerance, + ) + body = payload.decode("utf-8") if isinstance(payload, bytes) else payload + data = json.loads(body) + if not isinstance(data, dict): + _raise_deserialize_error( + "ActionContext", ValueError("Actions payload must be a JSON object") + ) + return cast(Dict[str, Any], data) + + +def _build_signed_response( + *, + action_type: ActionType, + verdict: Literal["Allow", "Deny"], + error_message: Optional[str], + secret: str, +) -> Dict[str, Any]: + timestamp = int(time.time() * 1000) + response_payload: Dict[str, Any] = { + "timestamp": timestamp, + "verdict": verdict, + } + if error_message is not None: + response_payload["error_message"] = error_message + + payload_json = json.dumps(response_payload, separators=(",", ":")) + signed_payload = f"{timestamp}.{payload_json}" + signature = _compute_signature(signed_payload, secret) + object_type = _ACTION_TYPE_TO_RESPONSE_OBJECT[action_type] + + return { + "object": object_type, + "payload": response_payload, + "signature": signature, + } + + +# --------------------------------------------------------------------------- +# Public helpers +# --------------------------------------------------------------------------- + + class Actions: - """AuthKit Actions request verification and response signing.""" + """AuthKit Actions request verification and response signing. + + These helpers are local and synchronous; they make no HTTP calls. + """ def verify_header( self, @@ -81,7 +552,17 @@ def verify_header( secret: str, tolerance: int = DEFAULT_TOLERANCE, ) -> None: - """Verify the signature of an Actions request.""" + """Verify the signature of an Actions request. + + Args: + payload: The raw request body (bytes or str), never a re-serialized dict. + sig_header: The value of the ``WorkOS-Signature`` request header. + secret: The Actions signing secret from the WorkOS dashboard. + tolerance: Maximum age of the request in seconds. Defaults to 30. + + Raises: + ValueError: If the signature is invalid or the timestamp is stale. + """ _verify_signature( payload=payload, sig_header=sig_header, @@ -89,71 +570,37 @@ def verify_header( tolerance=tolerance, ) - def construct_action( + def verify_action( self, *, payload: Union[bytes, str], sig_header: str, secret: str, tolerance: int = DEFAULT_TOLERANCE, - ) -> Dict[str, Any]: - """Verify and deserialize an Actions request payload.""" - self.verify_header( - payload=payload, - sig_header=sig_header, - secret=secret, - tolerance=tolerance, - ) - body = payload.decode("utf-8") if isinstance(payload, bytes) else payload - return json.loads(body) + ) -> ActionContext: + """Verify an Actions request and deserialize it into a typed context. - def sign_response( - self, - *, - action_type: ActionType, - verdict: Literal["Allow", "Deny"], - error_message: Optional[str] = None, - secret: str, - ) -> Dict[str, Any]: - """Build and sign an Actions response.""" - timestamp = int(time.time() * 1000) - response_payload: Dict[str, Any] = { - "timestamp": timestamp, - "verdict": verdict, - } - if error_message is not None: - response_payload["error_message"] = error_message - - payload_json = json.dumps(response_payload, separators=(",", ":")) - signed_payload = f"{timestamp}.{payload_json}" - signature = _compute_signature(signed_payload, secret) - object_type = _ACTION_TYPE_TO_RESPONSE_OBJECT[action_type] - - return { - "object": object_type, - "payload": response_payload, - "signature": signature, - } - - -class AsyncActions: - """Async variant of AuthKit Actions helpers.""" + Returns an :class:`AuthenticationActionContext` for a sign-in or a + :class:`UserRegistrationActionContext` for a sign-up; narrow with + ``isinstance``. Use :meth:`construct_action` for the raw payload instead. - def verify_header( - self, - *, - payload: Union[bytes, str], - sig_header: str, - secret: str, - tolerance: int = DEFAULT_TOLERANCE, - ) -> None: - """Verify the signature of an Actions request.""" - _verify_signature( + Args: + payload: The raw request body (bytes or str), never a re-serialized dict. + sig_header: The value of the ``WorkOS-Signature`` request header. + secret: The Actions signing secret from the WorkOS dashboard. + tolerance: Maximum age of the request in seconds. Defaults to 30. + + Raises: + ValueError: If the signature is invalid or the timestamp is stale. + WorkOSError: If the payload is not a recognized Actions request. + """ + data = _load_verified_payload( payload=payload, sig_header=sig_header, secret=secret, tolerance=tolerance, ) + return _deserialize_action_context(data) def construct_action( self, @@ -163,15 +610,20 @@ def construct_action( secret: str, tolerance: int = DEFAULT_TOLERANCE, ) -> Dict[str, Any]: - """Verify and deserialize an Actions request payload.""" - self.verify_header( + """Verify an Actions request and return its payload as a plain dict. + + Prefer :meth:`verify_action`, which returns a typed context. + + Raises: + ValueError: If the signature is invalid or the timestamp is stale. + WorkOSError: If the payload is not a JSON object. + """ + return _load_verified_payload( payload=payload, sig_header=sig_header, secret=secret, tolerance=tolerance, ) - body = payload.decode("utf-8") if isinstance(payload, bytes) else payload - return json.loads(body) def sign_response( self, @@ -181,22 +633,22 @@ def sign_response( error_message: Optional[str] = None, secret: str, ) -> Dict[str, Any]: - """Build and sign an Actions response.""" - timestamp = int(time.time() * 1000) - response_payload: Dict[str, Any] = { - "timestamp": timestamp, - "verdict": verdict, - } - if error_message is not None: - response_payload["error_message"] = error_message - - payload_json = json.dumps(response_payload, separators=(",", ":")) - signed_payload = f"{timestamp}.{payload_json}" - signature = _compute_signature(signed_payload, secret) - object_type = _ACTION_TYPE_TO_RESPONSE_OBJECT[action_type] - - return { - "object": object_type, - "payload": response_payload, - "signature": signature, - } + """Build and sign an Actions response. + + Returns the JSON-serializable body to respond with, including the + ``object``, ``payload`` and ``signature`` fields WorkOS expects. + """ + return _build_signed_response( + action_type=action_type, + verdict=verdict, + error_message=error_message, + secret=secret, + ) + + +class AsyncActions(Actions): + """AuthKit Actions helpers on the async client. + + Verification and signing are local computations, so the methods are the + same synchronous ones as :class:`Actions`. + """ diff --git a/tests/test_actions.py b/tests/test_actions.py index eb107905..87aefe13 100644 --- a/tests/test_actions.py +++ b/tests/test_actions.py @@ -1,10 +1,37 @@ +import copy import hashlib import hmac import json import time +from datetime import datetime, timezone +from typing import Any import pytest -from workos.actions import Actions, AsyncActions +from workos import WorkOSError +from workos.actions import ( + ActionInvitation, + ActionOrganization, + ActionOrganizationDomain, + ActionOrganizationMembership, + Actions, + ActionUser, + AsyncActions, + AuthenticationActionContext, + UserData, + UserRegistrationActionContext, +) +from workos.common.models.authenticate_response_authentication_method import ( + AuthenticateResponseAuthenticationMethod, +) +from workos.common.models.invitation_state import InvitationState +from workos.common.models.organization_domain_state import OrganizationDomainState +from workos.common.models.organization_membership_status import ( + OrganizationMembershipStatus, +) + +SECRET = "test_secret_key" +TIMESTAMP = "2024-10-22T17:12:50.746Z" +PARSED_TIMESTAMP = datetime(2024, 10, 22, 17, 12, 50, 746000, tzinfo=timezone.utc) def _make_sig_header(body: str, secret: str, timestamp_ms: int = 0) -> str: @@ -20,110 +47,463 @@ def _make_sig_header(body: str, secret: str, timestamp_ms: int = 0) -> str: return f"t={ts}, v1={sig}" -SAMPLE_ACTION_PAYLOAD = json.dumps( - { - "type": "authentication", - "user": {"id": "user_01", "email": "test@example.com"}, - "organization": {"id": "org_01"}, - } -) +def _sign(payload: dict[str, Any], secret: str = SECRET) -> tuple[str, str]: + """Serialize a payload and sign it the way WorkOS does.""" + body = json.dumps(payload) + return body, _make_sig_header(body, secret) -SECRET = "test_secret_key" +USER: dict[str, Any] = { + "object": "user", + "id": "user_01JATCHZVEC5EPANDPEZVM68Y9", + "email": "jane@foocorp.com", + "email_verified": True, + "first_name": "Jane", + "last_name": "Doe", + "name": "Jane Doe", + "profile_picture_url": "https://example.com/jane.jpg", + "last_sign_in_at": TIMESTAMP, + "locale": "en-US", + "created_at": TIMESTAMP, + "updated_at": TIMESTAMP, + "external_id": "ext_123", +} + +# What the API sends for a real sign-in: the full REST user and organization, +# plus the context-specific organization membership (primary role, no roles list). +AUTHENTICATION_PAYLOAD: dict[str, Any] = { + "object": "authentication_action_context", + "id": "actions_execution_01JATCMZJY26PQ59XT9BNT0FNN", + "issuer": "https://api.workos.com", + "user": USER, + "organization": { + "object": "organization", + "id": "org_01JATCMZJY26PQ59XT9BNT0FNN", + "name": "Foo Corp", + "allow_profiles_outside_organization": False, + "created_at": TIMESTAMP, + "updated_at": TIMESTAMP, + "domains": [ + { + "object": "organization_domain", + "id": "org_domain_01JATCNVYCHT1SZGENR4QTXKRM", + "organization_id": "org_01JATCMZJY26PQ59XT9BNT0FNN", + "domain": "foocorp.com", + "state": "verified", + "verification_strategy": "dns", + "verification_token": "tok_123", + "created_at": TIMESTAMP, + "updated_at": TIMESTAMP, + } + ], + "metadata": {"tier": "enterprise"}, + "stripe_customer_id": "cus_123", + "external_id": None, + }, + "organization_membership": { + "object": "organization_membership", + "id": "om_01JATCNVYCHT1SZGENR4QTXKRK", + "organization_id": "org_01JATCMZJY26PQ59XT9BNT0FNN", + "user_id": "user_01JATCHZVEC5EPANDPEZVM68Y9", + "status": "active", + "directory_managed": False, + "custom_attributes": {"department": "engineering"}, + "created_at": TIMESTAMP, + "updated_at": TIMESTAMP, + "role": {"slug": "member"}, + "user": USER, + }, + "ip_address": "50.141.123.10", + "user_agent": "Mozilla/5.0", + "device_fingerprint": "notafingerprint", + "authentication_method": "Password", + "client_id": "client_01JATCNVYCHT1SZGENR4QTXKRK", +} + +# What the dashboard's "send test request" sends: the API strips it down to the +# fields the actions-types contract guarantees, so the REST extras are absent. +AUTHENTICATION_TEST_PAYLOAD: dict[str, Any] = { + "object": "authentication_action_context", + "id": "actions_execution_01JATCMZJY26PQ59XT9BNT0FNM", + "issuer": "test", + "user": { + "object": "user", + "id": "user_01JATCHZVEC5EPANDPEZVM68Y9", + "email": "jane@foocorp.com", + "first_name": "Jane", + "last_name": "Doe", + "email_verified": True, + "profile_picture_url": "https://example.com/jane.jpg", + "created_at": TIMESTAMP, + "updated_at": TIMESTAMP, + }, + "organization": { + "object": "organization", + "id": "org_01JATCMZJY26PQ59XT9BNT0FNN", + "name": "Foo Corp", + "allow_profiles_outside_organization": False, + "domains": [], + "created_at": TIMESTAMP, + "updated_at": TIMESTAMP, + }, + "organization_membership": { + "object": "organization_membership", + "id": "om_01JATCNVYCHT1SZGENR4QTXKRK", + "user_id": "user_01JATCHZVEC5EPANDPEZVM68Y9", + "organization_id": "org_01JATCMZJY26PQ59XT9BNT0FNN", + "role": {"slug": "member"}, + "status": "active", + "created_at": TIMESTAMP, + "updated_at": TIMESTAMP, + }, + "ip_address": "50.141.123.10", + "user_agent": "Mozilla/5.0", + "authentication_method": "Password", +} + +USER_REGISTRATION_PAYLOAD: dict[str, Any] = { + "object": "user_registration_action_context", + "id": "actions_execution_01JATCMZJY26PQ59XT9BNT0FNP", + "user_data": { + "object": "user_data", + "email": "jane@foocorp.com", + "first_name": "Jane", + "last_name": "Doe", + "extra_attributes": {"company_size": "50"}, + }, + "invitation": { + "object": "invitation", + "id": "invitation_01JBVZWH8HJ855YZ5BWHG1WNZN", + "email": "jane@foocorp.com", + "inviter_user_id": "user_01JBVZWAEPWAE3YYKBVT0AF81F", + "organization_id": "org_01JBW46BTKAA98WZN8826XQ2YP", + "state": "accepted", + "accepted_at": "2024-10-22T17:13:50.746Z", + "revoked_at": None, + "expires_at": TIMESTAMP, + "created_at": "2024-10-21T17:12:50.746Z", + "updated_at": "2024-10-21T17:12:50.746Z", + }, + "ip_address": "50.141.123.10", + "user_agent": "Mozilla/5.0", + "device_fingerprint": "notafingerprint", + "authentication_method": "GoogleOAuth", + "client_id": "client_01JATCNVYCHT1SZGENR4QTXKRK", +} -class TestActions: +# A direct sign-up with no invitation and no names. +USER_REGISTRATION_PAYLOAD_MINIMAL: dict[str, Any] = { + "object": "user_registration_action_context", + "id": "actions_execution_01JATCMZJY26PQ59XT9BNT0FNQ", + "user_data": { + "object": "user_data", + "email": "sam@example.com", + "first_name": None, + "last_name": None, + }, + "authentication_method": "MagicAuth", + "client_id": "client_01JATCNVYCHT1SZGENR4QTXKRK", +} + + +class TestActionsVerifyHeader: def setup_method(self): self.actions = Actions() def test_verify_header_valid(self): - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET) - self.actions.verify_header( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header=sig, - secret=SECRET, - ) + body, sig = _sign(AUTHENTICATION_PAYLOAD) + self.actions.verify_header(payload=body, sig_header=sig, secret=SECRET) def test_verify_header_invalid_signature(self): - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET) + _, sig = _sign(AUTHENTICATION_PAYLOAD) with pytest.raises(ValueError, match="does not match"): self.actions.verify_header( - payload='{"tampered": true}', - sig_header=sig, - secret=SECRET, + payload='{"tampered": true}', sig_header=sig, secret=SECRET ) def test_verify_header_stale_timestamp(self): + body = json.dumps(AUTHENTICATION_PAYLOAD) old_ts = int((time.time() - 60) * 1000) - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET, old_ts) + sig = _make_sig_header(body, SECRET, old_ts) with pytest.raises(ValueError, match="tolerance zone"): self.actions.verify_header( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header=sig, - secret=SECRET, - tolerance=30, + payload=body, sig_header=sig, secret=SECRET, tolerance=30 ) def test_verify_header_future_timestamp(self): + body = json.dumps(AUTHENTICATION_PAYLOAD) future_ts = int((time.time() + 60) * 1000) - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET, future_ts) + sig = _make_sig_header(body, SECRET, future_ts) with pytest.raises(ValueError, match="tolerance zone"): self.actions.verify_header( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header=sig, - secret=SECRET, - tolerance=30, + payload=body, sig_header=sig, secret=SECRET, tolerance=30 ) def test_verify_header_custom_tolerance(self): + body = json.dumps(AUTHENTICATION_PAYLOAD) old_ts = int((time.time() - 10) * 1000) - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET, old_ts) + sig = _make_sig_header(body, SECRET, old_ts) self.actions.verify_header( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header=sig, - secret=SECRET, - tolerance=60, + payload=body, sig_header=sig, secret=SECRET, tolerance=60 ) def test_verify_header_malformed_header(self): + body = json.dumps(AUTHENTICATION_PAYLOAD) with pytest.raises(ValueError, match="Unable to extract"): self.actions.verify_header( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header="invalid-header", - secret=SECRET, + payload=body, sig_header="invalid-header", secret=SECRET ) def test_verify_header_bytes_payload(self): - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET) + body, sig = _sign(AUTHENTICATION_PAYLOAD) self.actions.verify_header( - payload=SAMPLE_ACTION_PAYLOAD.encode("utf-8"), - sig_header=sig, - secret=SECRET, + payload=body.encode("utf-8"), sig_header=sig, secret=SECRET ) - def test_construct_action_valid(self): - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET) + +class TestActionsVerifyAction: + def setup_method(self): + self.actions = Actions() + + def test_authentication_action_is_fully_typed(self): + body, sig = _sign(AUTHENTICATION_PAYLOAD) + action = self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + assert isinstance(action, AuthenticationActionContext) + assert action.object == "authentication_action_context" + assert action.id == "actions_execution_01JATCMZJY26PQ59XT9BNT0FNN" + assert action.issuer == "https://api.workos.com" + assert action.ip_address == "50.141.123.10" + assert action.user_agent == "Mozilla/5.0" + assert action.device_fingerprint == "notafingerprint" + assert action.client_id == "client_01JATCNVYCHT1SZGENR4QTXKRK" + assert ( + action.authentication_method + is AuthenticateResponseAuthenticationMethod.PASSWORD + ) + + assert isinstance(action.user, ActionUser) + assert action.user.id == "user_01JATCHZVEC5EPANDPEZVM68Y9" + assert action.user.email == "jane@foocorp.com" + assert action.user.email_verified is True + assert action.user.name == "Jane Doe" + assert action.user.external_id == "ext_123" + assert action.user.locale == "en-US" + assert action.user.last_sign_in_at == PARSED_TIMESTAMP + assert action.user.created_at == PARSED_TIMESTAMP + + assert isinstance(action.organization, ActionOrganization) + assert action.organization.id == "org_01JATCMZJY26PQ59XT9BNT0FNN" + assert action.organization.name == "Foo Corp" + assert action.organization.metadata == {"tier": "enterprise"} + assert action.organization.stripe_customer_id == "cus_123" + assert action.organization.external_id is None + assert action.organization.allow_profiles_outside_organization is False + assert len(action.organization.domains) == 1 + domain = action.organization.domains[0] + assert isinstance(domain, ActionOrganizationDomain) + assert domain.domain == "foocorp.com" + assert domain.state is OrganizationDomainState.VERIFIED + assert domain.verification_token == "tok_123" + assert domain.created_at == PARSED_TIMESTAMP + + membership = action.organization_membership + assert isinstance(membership, ActionOrganizationMembership) + assert membership.id == "om_01JATCNVYCHT1SZGENR4QTXKRK" + assert membership.role.slug == "member" + assert membership.status is OrganizationMembershipStatus.ACTIVE + assert membership.directory_managed is False + assert membership.custom_attributes == {"department": "engineering"} + assert isinstance(membership.user, ActionUser) + assert membership.user.id == action.user.id + + def test_authentication_action_accepts_dashboard_test_request(self): + body, sig = _sign(AUTHENTICATION_TEST_PAYLOAD) + action = self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + assert isinstance(action, AuthenticationActionContext) + assert action.issuer == "test" + assert action.device_fingerprint is None + assert action.client_id is None + assert action.user.first_name == "Jane" + assert action.user.name is None + assert action.user.external_id is None + assert action.user.last_sign_in_at is None + assert action.user.locale is None + assert action.user.metadata is None + assert isinstance(action.organization, ActionOrganization) + assert action.organization.domains == [] + assert action.organization.metadata is None + assert action.organization.external_id is None + assert isinstance(action.organization_membership, ActionOrganizationMembership) + assert action.organization_membership.directory_managed is None + assert action.organization_membership.custom_attributes is None + assert action.organization_membership.user is None + + def test_authentication_action_without_organization(self): + payload = copy.deepcopy(AUTHENTICATION_PAYLOAD) + del payload["organization"] + del payload["organization_membership"] + body, sig = _sign(payload) + action = self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + assert isinstance(action, AuthenticationActionContext) + assert action.organization is None + assert action.organization_membership is None + + def test_user_registration_action_is_fully_typed(self): + body, sig = _sign(USER_REGISTRATION_PAYLOAD) + action = self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + assert isinstance(action, UserRegistrationActionContext) + assert action.object == "user_registration_action_context" + assert action.id == "actions_execution_01JATCMZJY26PQ59XT9BNT0FNP" + assert ( + action.authentication_method + is AuthenticateResponseAuthenticationMethod.GOOGLE_OAUTH + ) + assert action.ip_address == "50.141.123.10" + assert action.client_id == "client_01JATCNVYCHT1SZGENR4QTXKRK" + + assert isinstance(action.user_data, UserData) + assert action.user_data.email == "jane@foocorp.com" + assert action.user_data.first_name == "Jane" + assert action.user_data.last_name == "Doe" + assert action.user_data.extra_attributes == {"company_size": "50"} + + assert isinstance(action.invitation, ActionInvitation) + assert action.invitation.id == "invitation_01JBVZWH8HJ855YZ5BWHG1WNZN" + assert action.invitation.state is InvitationState.ACCEPTED + assert action.invitation.organization_id == "org_01JBW46BTKAA98WZN8826XQ2YP" + assert action.invitation.inviter_user_id == "user_01JBVZWAEPWAE3YYKBVT0AF81F" + assert action.invitation.accepted_at == datetime( + 2024, 10, 22, 17, 13, 50, 746000, tzinfo=timezone.utc + ) + assert action.invitation.revoked_at is None + assert action.invitation.expires_at == PARSED_TIMESTAMP + + def test_user_registration_action_without_invitation(self): + body, sig = _sign(USER_REGISTRATION_PAYLOAD_MINIMAL) + action = self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + assert isinstance(action, UserRegistrationActionContext) + assert action.invitation is None + assert action.ip_address is None + assert action.user_agent is None + assert action.device_fingerprint is None + assert action.user_data.first_name is None + assert action.user_data.last_name is None + assert action.user_data.extra_attributes is None + assert ( + action.authentication_method + is AuthenticateResponseAuthenticationMethod.MAGIC_AUTH + ) + + def test_invitation_without_inviter_or_organization(self): + payload = copy.deepcopy(USER_REGISTRATION_PAYLOAD) + del payload["invitation"]["inviter_user_id"] + del payload["invitation"]["organization_id"] + body, sig = _sign(payload) + action = self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + assert isinstance(action, UserRegistrationActionContext) + assert isinstance(action.invitation, ActionInvitation) + assert action.invitation.inviter_user_id is None + assert action.invitation.organization_id is None + + def test_bytes_payload(self): + body, sig = _sign(AUTHENTICATION_PAYLOAD) + action = self.actions.verify_action( + payload=body.encode("utf-8"), sig_header=sig, secret=SECRET + ) + assert isinstance(action, AuthenticationActionContext) + + def test_unknown_authentication_method_is_preserved(self): + payload = copy.deepcopy(AUTHENTICATION_PAYLOAD) + payload["authentication_method"] = "SomeFutureOAuth" + body, sig = _sign(payload) + action = self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + assert isinstance(action, AuthenticationActionContext) + assert isinstance( + action.authentication_method, AuthenticateResponseAuthenticationMethod + ) + assert action.authentication_method.value == "SomeFutureOAuth" + + def test_invalid_signature_is_rejected_before_parsing(self): + payload = copy.deepcopy(AUTHENTICATION_PAYLOAD) + payload["object"] = "not_an_action" + body, sig = _sign(payload, secret="wrong_secret") + with pytest.raises(ValueError, match="does not match"): + self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + def test_unknown_object_raises_workos_error(self): + payload = copy.deepcopy(AUTHENTICATION_PAYLOAD) + payload["object"] = "mystery_action_context" + body, sig = _sign(payload) + with pytest.raises(WorkOSError, match="mystery_action_context"): + self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + def test_missing_required_field_raises_workos_error(self): + payload = copy.deepcopy(AUTHENTICATION_PAYLOAD) + del payload["user"]["email"] + body, sig = _sign(payload) + with pytest.raises(WorkOSError, match="ActionUser"): + self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + def test_non_object_payload_raises_workos_error(self): + body = "[]" + sig = _make_sig_header(body, SECRET) + with pytest.raises(WorkOSError, match="JSON object"): + self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + + def test_via_client(self, workos: Any): + body, sig = _sign(USER_REGISTRATION_PAYLOAD) + action = workos.actions.verify_action( + payload=body, sig_header=sig, secret=SECRET + ) + assert isinstance(action, UserRegistrationActionContext) + + @pytest.mark.asyncio + async def test_via_async_client(self, async_workos: Any): + body, sig = _sign(AUTHENTICATION_PAYLOAD) + action = async_workos.actions.verify_action( + payload=body, sig_header=sig, secret=SECRET + ) + assert isinstance(action, AuthenticationActionContext) + + +class TestActionsConstructAction: + def setup_method(self): + self.actions = Actions() + + def test_construct_action_returns_raw_dict(self): + body, sig = _sign(AUTHENTICATION_PAYLOAD) result = self.actions.construct_action( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header=sig, - secret=SECRET, + payload=body, sig_header=sig, secret=SECRET ) - assert result["type"] == "authentication" - assert result["user"]["id"] == "user_01" + assert result == AUTHENTICATION_PAYLOAD + assert result["user"]["id"] == "user_01JATCHZVEC5EPANDPEZVM68Y9" def test_construct_action_invalid_signature(self): - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, "wrong_secret") + body, sig = _sign(AUTHENTICATION_PAYLOAD, secret="wrong_secret") with pytest.raises(ValueError): - self.actions.construct_action( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header=sig, - secret=SECRET, - ) + self.actions.construct_action(payload=body, sig_header=sig, secret=SECRET) + + def test_construct_action_non_object_payload(self): + body = "[]" + sig = _make_sig_header(body, SECRET) + with pytest.raises(WorkOSError, match="JSON object"): + self.actions.construct_action(payload=body, sig_header=sig, secret=SECRET) + + +class TestActionsSignResponse: + def setup_method(self): + self.actions = Actions() def test_sign_response_authentication_allow(self): result = self.actions.sign_response( - action_type="authentication", - verdict="Allow", - secret=SECRET, + action_type="authentication", verdict="Allow", secret=SECRET ) assert result["object"] == "authentication_action_response" assert result["payload"]["verdict"] == "Allow" @@ -143,9 +523,7 @@ def test_sign_response_user_registration_deny(self): def test_sign_response_signature_is_verifiable(self): result = self.actions.sign_response( - action_type="authentication", - verdict="Allow", - secret=SECRET, + action_type="authentication", verdict="Allow", secret=SECRET ) ts = result["payload"]["timestamp"] payload_json = json.dumps(result["payload"], separators=(",", ":")) @@ -163,26 +541,24 @@ def setup_method(self): self.actions = AsyncActions() def test_verify_header_valid(self): - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET) - self.actions.verify_header( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header=sig, - secret=SECRET, - ) + body, sig = _sign(AUTHENTICATION_PAYLOAD) + self.actions.verify_header(payload=body, sig_header=sig, secret=SECRET) + + def test_verify_action_valid(self): + body, sig = _sign(USER_REGISTRATION_PAYLOAD) + action = self.actions.verify_action(payload=body, sig_header=sig, secret=SECRET) + assert isinstance(action, UserRegistrationActionContext) + assert action.user_data.email == "jane@foocorp.com" def test_construct_action_valid(self): - sig = _make_sig_header(SAMPLE_ACTION_PAYLOAD, SECRET) + body, sig = _sign(AUTHENTICATION_PAYLOAD) result = self.actions.construct_action( - payload=SAMPLE_ACTION_PAYLOAD, - sig_header=sig, - secret=SECRET, + payload=body, sig_header=sig, secret=SECRET ) - assert result["type"] == "authentication" + assert result["object"] == "authentication_action_context" def test_sign_response(self): result = self.actions.sign_response( - action_type="authentication", - verdict="Allow", - secret=SECRET, + action_type="authentication", verdict="Allow", secret=SECRET ) assert result["object"] == "authentication_action_response" From 32440ff9eb4df321c3d56bef5a9fc5a776782959 Mon Sep 17 00:00:00 2001 From: "Garen J. Torikian" Date: Tue, 10 Nov 2026 09:58:17 -0800 Subject: [PATCH 2/2] chore(actions): Describe the payload contract without naming an internal package --- src/workos/actions.py | 2 +- tests/test_actions.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/workos/actions.py b/src/workos/actions.py index 97c19875..598435a1 100644 --- a/src/workos/actions.py +++ b/src/workos/actions.py @@ -4,7 +4,7 @@ # are sent by WorkOS to the customer's endpoint and are not described by the # OpenAPI spec, so both the crypto and the payload models live here. # -# The payload models mirror the `@workos-inc/actions-types` contract in the +# The payload models mirror the Actions payload contract enforced by the # WorkOS API. Fields that contract guarantees are required; fields WorkOS adds # on a real sign-in (for example `user.external_id`) are optional, because a # test request sent from the dashboard omits them. The generated REST models diff --git a/tests/test_actions.py b/tests/test_actions.py index 87aefe13..613e8869 100644 --- a/tests/test_actions.py +++ b/tests/test_actions.py @@ -121,7 +121,7 @@ def _sign(payload: dict[str, Any], secret: str = SECRET) -> tuple[str, str]: } # What the dashboard's "send test request" sends: the API strips it down to the -# fields the actions-types contract guarantees, so the REST extras are absent. +# fields the Actions payload contract guarantees, so the REST extras are absent. AUTHENTICATION_TEST_PAYLOAD: dict[str, Any] = { "object": "authentication_action_context", "id": "actions_execution_01JATCMZJY26PQ59XT9BNT0FNM",