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..598435a1 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 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 +# (`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..613e8869 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 payload 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"