diff --git a/api.md b/api.md index 21359f56..11bd283d 100644 --- a/api.md +++ b/api.md @@ -150,8 +150,6 @@ Methods: - client.cards.renew(card_token, \*\*params) -> Card - client.cards.retrieve_spend_limits(card_token) -> CardSpendLimits - client.cards.search_by_pan(\*\*params) -> Card -- client.cards.get_embed_html(\*args) -> str -- client.cards.get_embed_url(\*args) -> URL ## AggregateBalances @@ -221,7 +219,6 @@ Methods: - client.disputes.initiate_evidence_upload(dispute_token, \*\*params) -> DisputeEvidence - client.disputes.list_evidences(dispute_token, \*\*params) -> SyncCursorPage[DisputeEvidence] - client.disputes.retrieve_evidence(evidence_token, \*, dispute_token) -> DisputeEvidence -- client.disputes.upload_evidence(\*args) -> None # Events @@ -236,7 +233,6 @@ Methods: - client.events.retrieve(event_token) -> Event - client.events.list(\*\*params) -> SyncCursorPage[Event] - client.events.list_attempts(event_token, \*\*params) -> SyncCursorPage[MessageAttempt] -- client.events.resend(\*args) -> None ## Subscriptions @@ -356,13 +352,6 @@ Methods: - client.responder_endpoints.delete(\*\*params) -> None - client.responder_endpoints.check_status(\*\*params) -> ResponderEndpointStatus -# Webhooks - -Methods: - -- client.webhooks.unwrap(\*args) -> object -- client.webhooks.verify_signature(\*args) -> None - # ExternalBankAccounts Types: diff --git a/examples/.keep b/examples/.keep new file mode 100644 index 00000000..d8c73e93 --- /dev/null +++ b/examples/.keep @@ -0,0 +1,4 @@ +File generated from our OpenAPI spec by Stainless. + +This directory can be used to store example files demonstrating usage of this SDK. +It is ignored by Stainless code generation and its content (other than this keep file) won't be touched. \ No newline at end of file diff --git a/integrations/pagination.py b/integrations/pagination.py deleted file mode 100755 index e874414f..00000000 --- a/integrations/pagination.py +++ /dev/null @@ -1,32 +0,0 @@ -#!/usr/bin/env -S rye run python integrations/pagination.py - -from __future__ import annotations - -import json - -from lithic import Lithic - -client = Lithic(environment="sandbox") - - -def main() -> None: - page = client.transactions.list() - assert len(page.data) > 0, "No transactions found" - - if not page.has_more or not page.has_next_page(): - raise RuntimeError(f"Expected multiple pages to be present, only got {len(page.data)} items") - - tokens: dict[str, int] = {} - - for transaction in page: - tokens[transaction.token] = tokens.get(transaction.token, 0) + 1 - - duplicates = {token: count for token, count in tokens.items() if count > 1} - if duplicates: - print(json.dumps(duplicates, indent=2)) # noqa: T201 - raise RuntimeError(f"Found {len(duplicates)} duplicate entries!") - - print("Success!") # noqa: T201 - - -main() diff --git a/src/lithic/_client.py b/src/lithic/_client.py index 3c0fda1a..dd58984a 100644 --- a/src/lithic/_client.py +++ b/src/lithic/_client.py @@ -76,7 +76,6 @@ class Lithic(SyncAPIClient): financial_accounts: resources.FinancialAccounts transactions: resources.Transactions responder_endpoints: resources.ResponderEndpoints - webhooks: resources.Webhooks external_bank_accounts: resources.ExternalBankAccounts payments: resources.Payments three_ds: resources.ThreeDS @@ -194,7 +193,6 @@ def __init__( self.financial_accounts = resources.FinancialAccounts(self) self.transactions = resources.Transactions(self) self.responder_endpoints = resources.ResponderEndpoints(self) - self.webhooks = resources.Webhooks(self) self.external_bank_accounts = resources.ExternalBankAccounts(self) self.payments = resources.Payments(self) self.three_ds = resources.ThreeDS(self) @@ -368,7 +366,6 @@ class AsyncLithic(AsyncAPIClient): financial_accounts: resources.AsyncFinancialAccounts transactions: resources.AsyncTransactions responder_endpoints: resources.AsyncResponderEndpoints - webhooks: resources.AsyncWebhooks external_bank_accounts: resources.AsyncExternalBankAccounts payments: resources.AsyncPayments three_ds: resources.AsyncThreeDS @@ -486,7 +483,6 @@ def __init__( self.financial_accounts = resources.AsyncFinancialAccounts(self) self.transactions = resources.AsyncTransactions(self) self.responder_endpoints = resources.AsyncResponderEndpoints(self) - self.webhooks = resources.AsyncWebhooks(self) self.external_bank_accounts = resources.AsyncExternalBankAccounts(self) self.payments = resources.AsyncPayments(self) self.three_ds = resources.AsyncThreeDS(self) diff --git a/src/lithic/resources/__init__.py b/src/lithic/resources/__init__.py index aabd710d..cb865d98 100644 --- a/src/lithic/resources/__init__.py +++ b/src/lithic/resources/__init__.py @@ -64,7 +64,6 @@ ThreeDSWithStreamingResponse, AsyncThreeDSWithStreamingResponse, ) -from .webhooks import Webhooks, AsyncWebhooks from .auth_rules import ( AuthRules, AsyncAuthRules, @@ -255,8 +254,6 @@ "AsyncResponderEndpointsWithRawResponse", "ResponderEndpointsWithStreamingResponse", "AsyncResponderEndpointsWithStreamingResponse", - "Webhooks", - "AsyncWebhooks", "ExternalBankAccounts", "AsyncExternalBankAccounts", "ExternalBankAccountsWithRawResponse", diff --git a/src/lithic/resources/cards/cards.py b/src/lithic/resources/cards/cards.py index c56972fb..ac1cbf44 100644 --- a/src/lithic/resources/cards/cards.py +++ b/src/lithic/resources/cards/cards.py @@ -2,16 +2,11 @@ from __future__ import annotations -import hmac -import json -import base64 -import hashlib from typing import Union -from datetime import datetime, timezone, timedelta +from datetime import datetime from typing_extensions import Literal import httpx -from httpx import URL from ... import _legacy_response from ...types import ( @@ -27,7 +22,6 @@ card_update_params, card_reissue_params, card_provision_params, - card_get_embed_url_params, card_search_by_pan_params, ) from ..._types import ( @@ -40,7 +34,6 @@ ) from ..._utils import ( maybe_transform, - strip_not_given, async_maybe_transform, ) from .balances import ( @@ -57,7 +50,6 @@ from ...pagination import SyncCursorPage, AsyncCursorPage from ..._base_client import ( AsyncPaginator, - _merge_mappings, make_request_options, ) from .aggregate_balances import ( @@ -541,117 +533,6 @@ def embed( cast_to=str, ) - def get_embed_html( - self, - *, - token: str, - css: str | NotGiven = NOT_GIVEN, - expiration: Union[str, datetime] | NotGiven = NOT_GIVEN, - target_origin: str | NotGiven = NOT_GIVEN, - # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. - # The extra values given here take precedence over values defined on the client or passed to this method. - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, - ) -> str: - """ - Generates and executes an embed request, returning html you can serve to the - user. - - Be aware that this html contains sensitive data whose presence on your server - could trigger PCI DSS. - - If your company is not certified PCI compliant, we recommend using - `get_embed_url()` instead. You would then pass that returned URL to the - frontend, where you can load it via an iframe. - """ - headers = _merge_mappings({"Accept": "text/html"}, extra_headers or {}) - url = self.get_embed_url( - css=css, - token=token, - expiration=expiration, - target_origin=target_origin, - ) - return self._get( - str(url), - cast_to=str, - options=make_request_options( - extra_headers=headers, - extra_query=extra_query, - extra_body=extra_body, - timeout=timeout, - ), - ) - - def get_embed_url( - self, - *, - token: str, - css: str | NotGiven = NOT_GIVEN, - expiration: Union[str, datetime] | NotGiven = NOT_GIVEN, - target_origin: str | NotGiven = NOT_GIVEN, - ) -> URL: - """ - Handling full card PANs and CVV codes requires that you comply with the Payment - Card Industry Data Security Standards (PCI DSS). Some clients choose to reduce - their compliance obligations by leveraging our embedded card UI solution - documented below. - - In this setup, PANs and CVV codes are presented to the end-user via a card UI - that we provide, optionally styled in the customer's branding using a specified - css stylesheet. A user's browser makes the request directly to api.lithic.com, - so card PANs and CVVs never touch the API customer's servers while full card - data is displayed to their end-users. The response contains an HTML document. - This means that the url for the request can be inserted straight into the `src` - attribute of an iframe. - - ```html - - ``` - - You should compute the request payload on the server side. You can render it (or - the whole iframe) on the server or make an ajax call from your front end code, - but **do not ever embed your API key into front end code, as doing so introduces - a serious security vulnerability**. - """ - # Default expiration of 1 minute from now. - if isinstance(expiration, NotGiven): - expiration = datetime.now(timezone.utc) + timedelta(minutes=1) - - query = maybe_transform( - strip_not_given( - { - "css": css, - "token": token, - "expiration": expiration, - "target_origin": target_origin, - } - ), - card_get_embed_url_params.CardGetEmbedURLParams, - ) - serialized = json.dumps(query, sort_keys=True, separators=(",", ":")) - params = { - "embed_request": base64.b64encode(bytes(serialized, "utf-8")).decode("utf-8"), - "hmac": base64.b64encode( - hmac.new( - key=bytes(self._client.api_key, "utf-8"), - msg=bytes(serialized, "utf-8"), - digestmod=hashlib.sha256, - ).digest() - ).decode("utf-8"), - } - - # Copied nearly directly from httpx.BaseClient._merge_url - base_url = self._client.base_url - raw_path = base_url.raw_path + URL("embed/card").raw_path - return base_url.copy_with(raw_path=raw_path).copy_merge_params(params) - def provision( self, card_token: str, @@ -1405,117 +1286,6 @@ async def embed( cast_to=str, ) - async def get_embed_html( - self, - *, - token: str, - css: str | NotGiven = NOT_GIVEN, - expiration: Union[str, datetime] | NotGiven = NOT_GIVEN, - target_origin: str | NotGiven = NOT_GIVEN, - # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. - # The extra values given here take precedence over values defined on the client or passed to this method. - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, - ) -> str: - """ - Generates and executes an embed request, returning html you can serve to the - user. - - Be aware that this html contains sensitive data whose presence on your server - could trigger PCI DSS. - - If your company is not certified PCI compliant, we recommend using - `get_embed_url()` instead. You would then pass that returned URL to the - frontend, where you can load it via an iframe. - """ - headers = _merge_mappings({"Accept": "text/html"}, extra_headers or {}) - url = self.get_embed_url( - css=css, - token=token, - expiration=expiration, - target_origin=target_origin, - ) - return await self._get( - str(url), - cast_to=str, - options=make_request_options( - extra_headers=headers, - extra_query=extra_query, - extra_body=extra_body, - timeout=timeout, - ), - ) - - def get_embed_url( - self, - *, - token: str, - css: str | NotGiven = NOT_GIVEN, - expiration: Union[str, datetime] | NotGiven = NOT_GIVEN, - target_origin: str | NotGiven = NOT_GIVEN, - ) -> URL: - """ - Handling full card PANs and CVV codes requires that you comply with the Payment - Card Industry Data Security Standards (PCI DSS). Some clients choose to reduce - their compliance obligations by leveraging our embedded card UI solution - documented below. - - In this setup, PANs and CVV codes are presented to the end-user via a card UI - that we provide, optionally styled in the customer's branding using a specified - css stylesheet. A user's browser makes the request directly to api.lithic.com, - so card PANs and CVVs never touch the API customer's servers while full card - data is displayed to their end-users. The response contains an HTML document. - This means that the url for the request can be inserted straight into the `src` - attribute of an iframe. - - ```html - - ``` - - You should compute the request payload on the server side. You can render it (or - the whole iframe) on the server or make an ajax call from your front end code, - but **do not ever embed your API key into front end code, as doing so introduces - a serious security vulnerability**. - """ - # Default expiration of 1 minute from now. - if isinstance(expiration, NotGiven): - expiration = datetime.now(timezone.utc) + timedelta(minutes=1) - - query = maybe_transform( - strip_not_given( - { - "css": css, - "token": token, - "expiration": expiration, - "target_origin": target_origin, - } - ), - card_get_embed_url_params.CardGetEmbedURLParams, - ) - serialized = json.dumps(query, sort_keys=True, separators=(",", ":")) - params = { - "embed_request": base64.b64encode(bytes(serialized, "utf-8")).decode("utf-8"), - "hmac": base64.b64encode( - hmac.new( - key=bytes(self._client.api_key, "utf-8"), - msg=bytes(serialized, "utf-8"), - digestmod=hashlib.sha256, - ).digest() - ).decode("utf-8"), - } - - # Copied nearly directly from httpx.BaseClient._merge_url - base_url = self._client.base_url - raw_path = base_url.raw_path + URL("embed/card").raw_path - return base_url.copy_with(raw_path=raw_path).copy_merge_params(params) - async def provision( self, card_token: str, diff --git a/src/lithic/resources/disputes.py b/src/lithic/resources/disputes.py index ef613e18..d2105ccc 100644 --- a/src/lithic/resources/disputes.py +++ b/src/lithic/resources/disputes.py @@ -18,16 +18,7 @@ dispute_list_evidences_params, dispute_initiate_evidence_upload_params, ) -from .._types import ( - NOT_GIVEN, - Body, - Omit, - Query, - Headers, - NoneType, - NotGiven, - FileTypes, -) +from .._types import NOT_GIVEN, Body, Query, Headers, NotGiven from .._utils import ( maybe_transform, async_maybe_transform, @@ -526,28 +517,6 @@ def retrieve_evidence( cast_to=DisputeEvidence, ) - def upload_evidence( - self, - dispute_token: str, - file: FileTypes, - *, - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - ) -> None: - """ - Initiates the Dispute Evidence Upload, then uploads the file to the returned - `upload_url`. - """ - payload = self._client.disputes.initiate_evidence_upload( - dispute_token, extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body - ) - if not payload.upload_url: - raise ValueError("Missing 'upload_url' from response payload") - files = {"file": file} - options = make_request_options(extra_headers={"Authorization": Omit()}) - self._put(payload.upload_url, cast_to=NoneType, body=None, files=files, options=options) - class AsyncDisputes(AsyncAPIResource): @cached_property @@ -1031,28 +1000,6 @@ async def retrieve_evidence( cast_to=DisputeEvidence, ) - async def upload_evidence( - self, - dispute_token: str, - file: FileTypes, - *, - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - ) -> None: - """ - Initiates the Dispute Evidence Upload, then uploads the file to the returned - `upload_url`. - """ - payload = await self._client.disputes.initiate_evidence_upload( - dispute_token, extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body - ) - if not payload.upload_url: - raise ValueError("Missing 'upload_url' from response payload") - files = {"file": file} - options = make_request_options(extra_headers={"Authorization": Omit()}) - await self._put(payload.upload_url, cast_to=NoneType, body=None, files=files, options=options) - class DisputesWithRawResponse: def __init__(self, disputes: Disputes) -> None: diff --git a/src/lithic/resources/events/events.py b/src/lithic/resources/events/events.py index 76bb29e4..a9d22cdb 100644 --- a/src/lithic/resources/events/events.py +++ b/src/lithic/resources/events/events.py @@ -9,13 +9,8 @@ import httpx from ... import _legacy_response -from ...types import ( - Event, - MessageAttempt, - event_list_params, - event_list_attempts_params, -) -from ..._types import NOT_GIVEN, Body, Query, Headers, NoneType, NotGiven +from ...types import Event, MessageAttempt, event_list_params, event_list_attempts_params +from ..._types import NOT_GIVEN, Body, Query, Headers, NotGiven from ..._utils import maybe_transform from ..._compat import cached_property from ..._resource import SyncAPIResource, AsyncAPIResource @@ -244,22 +239,6 @@ def list_attempts( model=MessageAttempt, ) - def resend( - self, - event_token: str, - *, - event_subscription_token: str, - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - ) -> None: - """Resend an event to an event subscription.""" - self._post( - f"/events/{event_token}/event_subscriptions/{event_subscription_token}/resend", - options=make_request_options(extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body), - cast_to=NoneType, - ) - class AsyncEvents(AsyncAPIResource): @cached_property @@ -468,22 +447,6 @@ def list_attempts( model=MessageAttempt, ) - async def resend( - self, - event_token: str, - *, - event_subscription_token: str, - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - ) -> None: - """Resend an event to an event subscription.""" - await self._post( - f"/events/{event_token}/event_subscriptions/{event_subscription_token}/resend", - options=make_request_options(extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body), - cast_to=NoneType, - ) - class EventsWithRawResponse: def __init__(self, events: Events) -> None: diff --git a/src/lithic/resources/webhooks.py b/src/lithic/resources/webhooks.py deleted file mode 100644 index 8b5094f8..00000000 --- a/src/lithic/resources/webhooks.py +++ /dev/null @@ -1,207 +0,0 @@ -# File generated from our OpenAPI spec by Stainless. - -from __future__ import annotations - -import hmac -import json -import math -import base64 -import hashlib -from datetime import datetime, timezone, timedelta - -from .._types import ( - HeadersLike, -) -from .._utils import ( - removeprefix, - get_required_header, -) -from .._resource import SyncAPIResource, AsyncAPIResource - -__all__ = ["Webhooks", "AsyncWebhooks"] - - -class Webhooks(SyncAPIResource): - def unwrap( - self, - payload: str | bytes, - headers: HeadersLike, - *, - secret: str | None = None, - ) -> object: - """Validates that the given payload was sent by Lithic and parses the payload.""" - self.verify_signature(payload=payload, headers=headers, secret=secret) - return json.loads(payload) - - def verify_signature( - self, - payload: str | bytes, - headers: HeadersLike, - *, - secret: str | None = None, - ) -> None: - """Validates whether or not the webhook payload was sent by Lithic. - - An error will be raised if the webhook payload was not sent by Lithic. - """ - if secret is None: - secret = self._client.webhook_secret - - if secret is None: - raise ValueError( - "The webhook secret must either be set using the env var, LITHIC_WEBHOOK_SECRET, on the client class, Lithic(webhook_secret='123'), or passed to this function" - ) - - try: - whsecret = base64.b64decode(removeprefix(secret, "whsec_")) - except Exception as err: - raise ValueError("Bad secret") from err - - msg_id = get_required_header(headers, "webhook-id") - msg_timestamp = get_required_header(headers, "webhook-timestamp") - - # validate the timestamp - webhook_tolerance = timedelta(minutes=5) - now = datetime.now(tz=timezone.utc) - - try: - timestamp = datetime.fromtimestamp(float(msg_timestamp), tz=timezone.utc) - except Exception as err: - raise ValueError("Invalid signature headers. Could not convert to timestamp") from err - - # too old - if timestamp < (now - webhook_tolerance): - raise ValueError("Webhook timestamp is too old") - - # too new - if timestamp > (now + webhook_tolerance): - raise ValueError("Webhook timestamp is too new") - - # create the signature - body = payload.decode("utf-8") if isinstance(payload, bytes) else payload - if not isinstance(body, str): # pyright: ignore[reportUnnecessaryIsInstance] - raise ValueError( - "Webhook body should be a string of JSON (or bytes which can be decoded to a utf-8 string), not a parsed dictionary." - ) - - timestamp_str = str(math.floor(timestamp.replace(tzinfo=timezone.utc).timestamp())) - - to_sign = f"{msg_id}.{timestamp_str}.{body}".encode() - expected_signature = hmac.new(whsecret, to_sign, hashlib.sha256).digest() - - msg_signature = get_required_header(headers, "webhook-signature") - - # Signature header can contain multiple signatures delimited by spaces - passed_sigs = msg_signature.split(" ") - - for versioned_sig in passed_sigs: - values = versioned_sig.split(",") - if len(values) != 2: - # signature is not formatted like {version},{signature} - continue - - (version, signature) = values - - # Only verify prefix v1 - if version != "v1": - continue - - sig_bytes = base64.b64decode(signature) - if hmac.compare_digest(expected_signature, sig_bytes): - # valid! - return None - - raise ValueError("None of the given webhook signatures match the expected signature") - - -class AsyncWebhooks(AsyncAPIResource): - def unwrap( - self, - payload: str | bytes, - headers: HeadersLike, - *, - secret: str | None = None, - ) -> object: - """Validates that the given payload was sent by Lithic and parses the payload.""" - self.verify_signature(payload=payload, headers=headers, secret=secret) - return json.loads(payload) - - def verify_signature( - self, - payload: str | bytes, - headers: HeadersLike, - *, - secret: str | None = None, - ) -> None: - """Validates whether or not the webhook payload was sent by Lithic. - - An error will be raised if the webhook payload was not sent by Lithic. - """ - if secret is None: - secret = self._client.webhook_secret - - if secret is None: - raise ValueError( - "The webhook secret must either be set using the env var, LITHIC_WEBHOOK_SECRET, on the client class, Lithic(webhook_secret='123'), or passed to this function" - ) - - try: - whsecret = base64.b64decode(removeprefix(secret, "whsec_")) - except Exception as err: - raise ValueError("Bad secret") from err - - msg_id = get_required_header(headers, "webhook-id") - msg_timestamp = get_required_header(headers, "webhook-timestamp") - - # validate the timestamp - webhook_tolerance = timedelta(minutes=5) - now = datetime.now(tz=timezone.utc) - - try: - timestamp = datetime.fromtimestamp(float(msg_timestamp), tz=timezone.utc) - except Exception as err: - raise ValueError("Invalid signature headers. Could not convert to timestamp") from err - - # too old - if timestamp < (now - webhook_tolerance): - raise ValueError("Webhook timestamp is too old") - - # too new - if timestamp > (now + webhook_tolerance): - raise ValueError("Webhook timestamp is too new") - - # create the signature - body = payload.decode("utf-8") if isinstance(payload, bytes) else payload - if not isinstance(body, str): # pyright: ignore[reportUnnecessaryIsInstance] - raise ValueError( - "Webhook body should be a string of JSON (or bytes which can be decoded to a utf-8 string), not a parsed dictionary." - ) - - timestamp_str = str(math.floor(timestamp.replace(tzinfo=timezone.utc).timestamp())) - - to_sign = f"{msg_id}.{timestamp_str}.{body}".encode() - expected_signature = hmac.new(whsecret, to_sign, hashlib.sha256).digest() - - msg_signature = get_required_header(headers, "webhook-signature") - - # Signature header can contain multiple signatures delimited by spaces - passed_sigs = msg_signature.split(" ") - - for versioned_sig in passed_sigs: - values = versioned_sig.split(",") - if len(values) != 2: - # signature is not formatted like {version},{signature} - continue - - (version, signature) = values - - # Only verify prefix v1 - if version != "v1": - continue - - sig_bytes = base64.b64decode(signature) - if hmac.compare_digest(expected_signature, sig_bytes): - # valid! - return None - - raise ValueError("None of the given webhook signatures match the expected signature") diff --git a/src/lithic/types/__init__.py b/src/lithic/types/__init__.py index bc1b39f6..fce1f10f 100644 --- a/src/lithic/types/__init__.py +++ b/src/lithic/types/__init__.py @@ -38,7 +38,6 @@ from .card_embed_response import CardEmbedResponse as CardEmbedResponse from .card_reissue_params import CardReissueParams as CardReissueParams from .dispute_list_params import DisputeListParams as DisputeListParams -from .event_resend_params import EventResendParams as EventResendParams from .payment_list_params import PaymentListParams as PaymentListParams from .tokenization_secret import TokenizationSecret as TokenizationSecret from .verification_method import VerificationMethod as VerificationMethod @@ -63,11 +62,9 @@ from .card_program_list_params import CardProgramListParams as CardProgramListParams from .tokenization_list_params import TokenizationListParams as TokenizationListParams from .auth_rule_remove_response import AuthRuleRemoveResponse as AuthRuleRemoveResponse -from .card_get_embed_url_params import CardGetEmbedURLParams as CardGetEmbedURLParams from .card_search_by_pan_params import CardSearchByPanParams as CardSearchByPanParams from .responder_endpoint_status import ResponderEndpointStatus as ResponderEndpointStatus from .account_holder_list_params import AccountHolderListParams as AccountHolderListParams -from .card_get_embed_html_params import CardGetEmbedHTMLParams as CardGetEmbedHTMLParams from .event_list_attempts_params import EventListAttemptsParams as EventListAttemptsParams from .settlement_summary_details import SettlementSummaryDetails as SettlementSummaryDetails from .auth_rule_retrieve_response import AuthRuleRetrieveResponse as AuthRuleRetrieveResponse diff --git a/src/lithic/types/card_get_embed_html_params.py b/src/lithic/types/card_get_embed_html_params.py deleted file mode 100644 index 86bb86d3..00000000 --- a/src/lithic/types/card_get_embed_html_params.py +++ /dev/null @@ -1,41 +0,0 @@ -# File generated from our OpenAPI spec by Stainless. - -from __future__ import annotations - -from typing import Union -from datetime import datetime -from typing_extensions import Required, Annotated, TypedDict - -from .._utils import PropertyInfo - -__all__ = ["CardGetEmbedHTMLParams"] - - -class CardGetEmbedHTMLParams(TypedDict, total=False): - token: Required[str] - """Globally unique identifier for the card to be displayed.""" - - css: str - """ - A publicly available URI, so the white-labeled card element can be styled with - the client's branding. - """ - - expiration: Annotated[Union[str, datetime], PropertyInfo(format="iso8601")] - """An RFC 3339 timestamp for when the request should expire. UTC time zone. - - If no timezone is specified, UTC will be used. If payload does not contain an - expiration, the request will never expire. - - Using an `expiration` reduces the risk of a - [replay attack](https://en.wikipedia.org/wiki/Replay_attack). Without supplying - the `expiration`, in the event that a malicious user gets a copy of your request - in transit, they will be able to obtain the response data indefinitely. - """ - - target_origin: str - """Required if you want to post the element clicked to the parent iframe. - - If you supply this param, you can also capture click events in the parent iframe - by adding an event listener. - """ diff --git a/src/lithic/types/card_get_embed_url_params.py b/src/lithic/types/card_get_embed_url_params.py deleted file mode 100644 index df77c553..00000000 --- a/src/lithic/types/card_get_embed_url_params.py +++ /dev/null @@ -1,41 +0,0 @@ -# File generated from our OpenAPI spec by Stainless. - -from __future__ import annotations - -from typing import Union -from datetime import datetime -from typing_extensions import Required, Annotated, TypedDict - -from .._utils import PropertyInfo - -__all__ = ["CardGetEmbedURLParams"] - - -class CardGetEmbedURLParams(TypedDict, total=False): - token: Required[str] - """Globally unique identifier for the card to be displayed.""" - - css: str - """ - A publicly available URI, so the white-labeled card element can be styled with - the client's branding. - """ - - expiration: Annotated[Union[str, datetime], PropertyInfo(format="iso8601")] - """An RFC 3339 timestamp for when the request should expire. UTC time zone. - - If no timezone is specified, UTC will be used. If payload does not contain an - expiration, the request will never expire. - - Using an `expiration` reduces the risk of a - [replay attack](https://en.wikipedia.org/wiki/Replay_attack). Without supplying - the `expiration`, in the event that a malicious user gets a copy of your request - in transit, they will be able to obtain the response data indefinitely. - """ - - target_origin: str - """Required if you want to post the element clicked to the parent iframe. - - If you supply this param, you can also capture click events in the parent iframe - by adding an event listener. - """ diff --git a/src/lithic/types/event_resend_params.py b/src/lithic/types/event_resend_params.py deleted file mode 100644 index e381da50..00000000 --- a/src/lithic/types/event_resend_params.py +++ /dev/null @@ -1,41 +0,0 @@ -# File generated from our OpenAPI spec by Stainless. - -from __future__ import annotations - -from typing import Union -from datetime import datetime -from typing_extensions import Required, Annotated, TypedDict - -from .._utils import PropertyInfo - -__all__ = ["EventResendParams"] - - -class EventResendParams(TypedDict, total=False): - token: Required[str] - """Globally unique identifier for the card to be displayed.""" - - css: str - """ - A publicly available URI, so the white-labeled card element can be styled with - the client's branding. - """ - - expiration: Annotated[Union[str, datetime], PropertyInfo(format="iso8601")] - """An RFC 3339 timestamp for when the request should expire. UTC time zone. - - If no timezone is specified, UTC will be used. If payload does not contain an - expiration, the request will never expire. - - Using an `expiration` reduces the risk of a - [replay attack](https://en.wikipedia.org/wiki/Replay_attack). Without supplying - the `expiration`, in the event that a malicious user gets a copy of your request - in transit, they will be able to obtain the response data indefinitely. - """ - - target_origin: str - """Required if you want to post the element clicked to the parent iframe. - - If you supply this param, you can also capture click events in the parent iframe - by adding an event listener. - """ diff --git a/tests/api_resources/test_cards.py b/tests/api_resources/test_cards.py index a01331e0..fa03d2ec 100644 --- a/tests/api_resources/test_cards.py +++ b/tests/api_resources/test_cards.py @@ -250,18 +250,6 @@ def test_streaming_response_embed(self, client: Lithic) -> None: assert cast(Any, response.is_closed) is True - def test_get_embed_html(self, client: Lithic) -> None: - html = client.cards.get_embed_html(token="foo") - assert "html" in html - - def test_get_embed_url(self, client: Lithic) -> None: - url = client.cards.get_embed_url(token="foo") - params = set( # pyright: ignore[reportUnknownVariableType] - url.params.keys() # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType] - ) - assert "hmac" in params - assert "embed_request" in params - @parametrize def test_method_provision(self, client: Lithic) -> None: card = client.cards.provision( @@ -771,18 +759,6 @@ async def test_streaming_response_embed(self, async_client: AsyncLithic) -> None assert cast(Any, response.is_closed) is True - async def test_get_embed_html(self, async_client: AsyncLithic) -> None: - html = await async_client.cards.get_embed_html(token="foo") - assert "html" in html - - def test_get_embed_url(self, async_client: Lithic) -> None: - url = async_client.cards.get_embed_url(token="foo") - params = set( # pyright: ignore[reportUnknownVariableType] - url.params.keys() # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType] - ) - assert "hmac" in params - assert "embed_request" in params - @parametrize async def test_method_provision(self, async_client: AsyncLithic) -> None: card = await async_client.cards.provision( diff --git a/tests/api_resources/test_events.py b/tests/api_resources/test_events.py index 893dc631..267caa8c 100644 --- a/tests/api_resources/test_events.py +++ b/tests/api_resources/test_events.py @@ -146,13 +146,6 @@ def test_path_params_list_attempts(self, client: Lithic) -> None: "", ) - @pytest.mark.skip(reason="Prism Mock server doesnt want Accept header, but server requires it.") - def test_method_resend(self, client: Lithic) -> None: - client.events.resend( - "string", - event_subscription_token="string", - ) - class TestAsyncEvents: parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"]) @@ -283,10 +276,3 @@ async def test_path_params_list_attempts(self, async_client: AsyncLithic) -> Non await async_client.events.with_raw_response.list_attempts( "", ) - - @pytest.mark.skip(reason="Prism Mock server doesnt want Accept header, but server requires it.") - async def test_method_resend(self, async_client: AsyncLithic) -> None: - await async_client.events.resend( - "string", - event_subscription_token="string", - ) diff --git a/tests/api_resources/test_webhooks.py b/tests/api_resources/test_webhooks.py deleted file mode 100644 index c3fc25b5..00000000 --- a/tests/api_resources/test_webhooks.py +++ /dev/null @@ -1,211 +0,0 @@ -# File generated from our OpenAPI spec by Stainless. - -from __future__ import annotations - -import os -import base64 -from typing import Any, cast -from datetime import datetime, timezone, timedelta - -import pytest -import time_machine - -from lithic import Lithic, AsyncLithic - -base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") - - -class TestWebhooks: - parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) - - timestamp = "1676312382" - fake_now = datetime.fromtimestamp(float(timestamp), tz=timezone.utc) - - payload = """{"card_token":"sit Lorem ipsum, accusantium repellendus possimus","created_at":"elit. placeat libero architecto molestias, sit","account_token":"elit.","issuer_decision":"magnam, libero esse Lorem ipsum magnam, magnam,","tokenization_attempt_id":"illum dolor repellendus libero esse accusantium","wallet_decisioning_info":{"device_score":"placeat architecto"},"digital_wallet_token_metadata":{"status":"reprehenderit dolor","token_requestor_id":"possimus","payment_account_info":{"account_holder_data":{"phone_number":"libero","email_address":"nobis molestias, veniam culpa! quas elit. quas libero esse architecto placeat"},"pan_unique_reference":"adipisicing odit magnam, odit"}}}""" - signature = "Dwa0AHInLL3XFo2sxcHamOQDrJNi7F654S3L6skMAOI=" - headers = { - "webhook-id": "msg_2Lh9KRb0pzN4LePd3XiA4v12Axj", - "webhook-timestamp": timestamp, - "webhook-signature": f"v1,{signature}", - } - secret = "whsec_zlFsbBZ8Xcodlpcu6NDTdSzZRLSdhkst" - - @time_machine.travel(fake_now) - def test_unwrap(self, client: Lithic) -> None: - payload = self.payload - headers = self.headers - secret = self.secret - - client.webhooks.unwrap(payload, headers, secret=secret) - - @time_machine.travel(fake_now) - def test_verify_signature(self, client: Lithic) -> None: - payload = self.payload - headers = self.headers - secret = self.secret - signature = self.signature - verify = client.webhooks.verify_signature - - assert verify(payload=payload, headers=headers, secret=secret) is None - - with pytest.raises(ValueError, match="Webhook timestamp is too old"): - with time_machine.travel(self.fake_now + timedelta(minutes=6)): - assert verify(payload=payload, headers=headers, secret=secret) is False - - with pytest.raises(ValueError, match="Webhook timestamp is too new"): - with time_machine.travel(self.fake_now - timedelta(minutes=6)): - assert verify(payload=payload, headers=headers, secret=secret) is False - - # wrong secret - with pytest.raises(ValueError, match=r"Bad secret"): - verify(payload=payload, headers=headers, secret="invalid secret") - - invalid_signature_message = "None of the given webhook signatures match the expected signature" - with pytest.raises(ValueError, match=invalid_signature_message): - verify( - payload=payload, - headers=headers, - secret=f"whsec_{base64.b64encode(b'foo').decode('utf-8')}", - ) - - # multiple signatures - invalid_signature = base64.b64encode(b"my_sig").decode("utf-8") - assert ( - verify( - payload=payload, - headers={**headers, "webhook-signature": f"v1,{invalid_signature} v1,{signature}"}, - secret=secret, - ) - is None - ) - - # different signature version - with pytest.raises(ValueError, match=invalid_signature_message): - verify( - payload=payload, - headers={**headers, "webhook-signature": f"v2,{signature}"}, - secret=secret, - ) - - assert ( - verify( - payload=payload, - headers={**headers, "webhook-signature": f"v2,{signature} v1,{signature}"}, - secret=secret, - ) - is None - ) - - # missing version - with pytest.raises(ValueError, match=invalid_signature_message): - verify( - payload=payload, - headers={**headers, "webhook-signature": signature}, - secret=secret, - ) - - # non-string payload - with pytest.raises(ValueError, match=r"Webhook body should be a string"): - verify( - payload=cast(Any, {"payload": payload}), - headers=headers, - secret=secret, - ) - - -class TestAsyncWebhooks: - parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"]) - - timestamp = "1676312382" - fake_now = datetime.fromtimestamp(float(timestamp), tz=timezone.utc) - - payload = """{"card_token":"sit Lorem ipsum, accusantium repellendus possimus","created_at":"elit. placeat libero architecto molestias, sit","account_token":"elit.","issuer_decision":"magnam, libero esse Lorem ipsum magnam, magnam,","tokenization_attempt_id":"illum dolor repellendus libero esse accusantium","wallet_decisioning_info":{"device_score":"placeat architecto"},"digital_wallet_token_metadata":{"status":"reprehenderit dolor","token_requestor_id":"possimus","payment_account_info":{"account_holder_data":{"phone_number":"libero","email_address":"nobis molestias, veniam culpa! quas elit. quas libero esse architecto placeat"},"pan_unique_reference":"adipisicing odit magnam, odit"}}}""" - signature = "Dwa0AHInLL3XFo2sxcHamOQDrJNi7F654S3L6skMAOI=" - headers = { - "webhook-id": "msg_2Lh9KRb0pzN4LePd3XiA4v12Axj", - "webhook-timestamp": timestamp, - "webhook-signature": f"v1,{signature}", - } - secret = "whsec_zlFsbBZ8Xcodlpcu6NDTdSzZRLSdhkst" - - @time_machine.travel(fake_now) - def test_unwrap(self, async_client: AsyncLithic) -> None: - payload = self.payload - headers = self.headers - secret = self.secret - - async_client.webhooks.unwrap(payload, headers, secret=secret) - - @time_machine.travel(fake_now) - def test_verify_signature(self, async_client: Lithic) -> None: - payload = self.payload - headers = self.headers - secret = self.secret - signature = self.signature - verify = async_client.webhooks.verify_signature - - assert verify(payload=payload, headers=headers, secret=secret) is None - - with pytest.raises(ValueError, match="Webhook timestamp is too old"): - with time_machine.travel(self.fake_now + timedelta(minutes=6)): - assert verify(payload=payload, headers=headers, secret=secret) is False - - with pytest.raises(ValueError, match="Webhook timestamp is too new"): - with time_machine.travel(self.fake_now - timedelta(minutes=6)): - assert verify(payload=payload, headers=headers, secret=secret) is False - - # wrong secret - with pytest.raises(ValueError, match=r"Bad secret"): - verify(payload=payload, headers=headers, secret="invalid secret") - - invalid_signature_message = "None of the given webhook signatures match the expected signature" - with pytest.raises(ValueError, match=invalid_signature_message): - verify( - payload=payload, - headers=headers, - secret=f"whsec_{base64.b64encode(b'foo').decode('utf-8')}", - ) - - # multiple signatures - invalid_signature = base64.b64encode(b"my_sig").decode("utf-8") - assert ( - verify( - payload=payload, - headers={**headers, "webhook-signature": f"v1,{invalid_signature} v1,{signature}"}, - secret=secret, - ) - is None - ) - - # different signature version - with pytest.raises(ValueError, match=invalid_signature_message): - verify( - payload=payload, - headers={**headers, "webhook-signature": f"v2,{signature}"}, - secret=secret, - ) - - assert ( - verify( - payload=payload, - headers={**headers, "webhook-signature": f"v2,{signature} v1,{signature}"}, - secret=secret, - ) - is None - ) - - # missing version - with pytest.raises(ValueError, match=invalid_signature_message): - verify( - payload=payload, - headers={**headers, "webhook-signature": signature}, - secret=secret, - ) - - # non-string payload - with pytest.raises(ValueError, match=r"Webhook body should be a string"): - verify( - payload=cast(Any, {"payload": payload}), - headers=headers, - secret=secret, - )