CVE-2023-32694 (Python, published 2023-05-25T15:15Z)
Description: value: Saleor Core is a composable, headless commerce API. Saleor's `validate_hmac_signature` function is vulnerable to timing attacks. Malicious users could abuse this vulnerability on Saleor deployments having the Adyen plugin enabled in order to determine the secret key and forge fake events, this could affect the database integrity such as marking an order as paid when it is not. This issue has been patched in versions 3.7.68, 3.8.40, 3.9.49, 3.10.36, 3.11.35, 3.12.25, and 3.13.16.
Severity: nan (CVSS v3 base score: 5.4)
Weakness: CWE-203 \u2014 Observable Discrepancy
The product behaves differently or sends different responses under different circumstances in a way that is observable to an unauthorized actor, which exposes security-relevant information about the state of the product, such as whether a particular operation was successful or not.
Repository: https://github.com/saleor/saleor
Files changed:
saleor/graphql/schema.graphql
saleor/order/__init__.py
saleor/order/events.py
saleor/payment/__init__.py
saleor/payment/gateways/adyen/plugin.py
saleor/payment/gateways/adyen/utils.py
saleor/payment/gateways/adyen/webhooks.py
saleor/payment/migrations/0019_auto_20200721_1313.py
saleor/payment/utils.py
Commit message: Add webhooks logic


Vulnerable code:
  checkoutComplete(checkoutId: ID!, redirectUrl: String, storeSource: Boolean = false): CheckoutComplete
  paymentData: JSONString
from typing import Optional
from django.http import HttpResponse, JsonResponse
        {"name": "Enable notifications", "value": True},
                "webhook url is <your-backend-url>/plugins/mirumee.payments.adyen/"
                "webhooks/ "
                "ions-in-your-customer-area"
                "set-up-notifications-in-your-customer-area"
                "set-up-notifications-in-your-customer-area"
            auto_capture=True,  # FIXME check this

        print(request.body)
        return HttpResponse("[accepted]")
    payment_information: PaymentData,
    if transaction_kind in {TransactionKind.CAPTURE, TransactionKind.CONFIRM}:

Fixed code:
  checkoutComplete(checkoutId: ID!, paymentData: JSONString, redirectUrl: String, storeSource: Boolean = false): CheckoutComplete
  PAYMENT_GATEWAY_NOTIFICATION
  PENDING
  REFUSED
  CANCELLED
  PENDING
  REFUND_ONGOING
  CANCEL
    PAYMENT_GATEWAY_NOTIFICATION = "payment_gateway_notification"
        (PAYMENT_GATEWAY_NOTIFICATION, "Notification from payment gateway"),
def payment_gateway_notification_event(
    *, order: Order, user: UserType, message: str, payment: Payment
) -> OrderEvent:
    if not _user_is_valid(user):
        user = None
    parameters = {"message": message}

    if payment:
        parameters.update({"gateway": payment.gateway, "payment_id": payment.token})

    return OrderEvent.objects.create(
        order=order,
        type=OrderEvents.PAYMENT_GATEWAY_NOTIFICATION,
        user=user,
        parameters=parameters,
    )


    CAPTURE_FAILED = "capture_failed"
    REFUND_FAILED = "refund_failed"
    REFUND_REVERSED = "refund_reversed"
    CANCEL = "cancel"
        (CANCEL, "Cancel"),
    REFUSED = "refused"
    CANCELLED = "cancelled"
        (REFUSED, "Refused"),
        (CANCELLED, "Cancelled"),
from typing import List, Optional
from django.contrib.auth.hashers import make_password
from django.http import HttpResponse, HttpResponseNotFound, JsonResponse
from .webhooks import handle_webhook
                "webhook url is http(s)://<your-backend-url>/plugins/mirumee.payments.adyen/webhooks/ "
                "ions-in-your-customer-area. The Saleor webhook url is "
                "http(s)://<your-backend-url>/plugins/mirumee.payments.adyen/webhooks/"
                "set-up-notifications-in-your-customer-area "
                "The Saleor webhook url is "
                "http(s)://<your-backend-url>/plugins/mirumee.payments.adyen/webhooks/"
                "set-up-notifications-in-your-customer-area "
                "The Saleor webhook url is "
                "http(s)://<your-backend-url>/plugins/mirumee.payments.adyen/webhooks/"
            auto_capture=configuration[
                "Automatically mark payment as a capture"
            ],  # FIXME check this
                "live": configuration["Live"],
                "webhook_hmac": configuration["HMAC secret key"],
                "webhook_user": configuration["Notification user"],
                "webhook_user_password": configuration["Notification password"],
        config = self._get_gateway_config()
        return handle_webhook(request, config)
    @classmethod
    def _update_config_items(
        cls, configuration_to_update: List[dict], current_config: List[dict]
    ):
        super()._update_config_items(configuration_to_update, current_config)
        for item in current_config:
            if item.get("name") == "Notification password":
                item["value"] = make_password(item["value"])

def convert_adyen_price_format(value: str, currency: str):
    value = Decimal(value)
    precision = get_currency_precision(currency)
    number_places = Decimal(10) ** -precision
    return value * number_places


import base64
import binascii
import hashlib
import hmac
import json
from typing import Any, Dict, Optional

from django.contrib.auth.hashers import check_password
from django.core.handlers.wsgi import WSGIRequest
from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseNotFound
from django.http.request import HttpHeaders
from graphql_relay import from_global_id

from ....order.events import payment_gateway_notification_event
from ....payment.models import Payment, Transaction
from ... import TransactionKind
from ...interface import GatewayConfig, GatewayResponse
from ...utils import create_transaction, gateway_postprocess
from .utils import convert_adyen_price_format


def get_payment(payment_id: str) -> Payment:
    _type, payment_id = from_global_id(payment_id)
    payment = Payment.objects.prefetch_related("order").filter(id=payment_id).first()
    return payment


def get_transaction(
    payment: "Payment", transaction_id: str, kind: TransactionKind,
) -> Transaction:
    transaction = payment.transactions.filter(kind=kind, token=transaction_id)
    return transaction


def create_new_transaction(notification, payment, kind):
    transaction_id = notification.get("pspReference")
    currency = notification.get("amount", {}).get("currency")
    amount = convert_adyen_price_format(
        notification.get("amount", {}).get("value"), currency
    )
    is_success = True if notification.get("success") == "true" else False

    gateway_response = GatewayResponse(
        kind=kind,
        action_required=False,
        transaction_id=transaction_id,
        is_success=is_success,
        amount=amount,
        currency=currency,
        error="",
        raw_response={},
    )
    return create_transaction(
        payment,
        kind=kind,
        payment_information=None,
        action_required=False,
        gateway_response=gateway_response,
    )


def create_payment_notification_for_order(
    payment: Payment, success_msg: str, failed_msg: Optional[str], is_success: bool
):
    if not payment.order:
        # Order is not assigned
        return
    msg = success_msg if is_success else failed_msg

    payment_gateway_notification_event(
        order=payment.order, user=None, message=msg, payment=payment
    )


def handle_authorization(notification: Dict[str, Any], gateway_config: GatewayConfig):
    mark_capture = gateway_config.auto_capture
    if mark_capture:
        # If we mark order as a capture by default we don't need to handle auth actions
        return
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        # We don't know anything about that payment
        return

    transaction_id = notification.get("pspReference")
    transaction = get_transaction(payment, transaction_id, TransactionKind.AUTH)
    if transaction:
        # We already marked it as Auth
        return

    transaction = create_new_transaction(notification, payment, TransactionKind.AUTH)

    success_msg = "Adyen: The payment request was successful."
    failed_msg = "Adyen: The payment request failed."
    create_payment_notification_for_order(
        payment, success_msg, failed_msg, transaction.is_success
    )


def handle_cancellation(notification: Dict[str, Any], _gateway_config: GatewayConfig):
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    transaction_id = notification.get("pspReference")
    transaction = get_transaction(payment, transaction_id, TransactionKind.CANCEL)
    if transaction:
        # it is already cancelled
        return
    new_transaction = create_new_transaction(
        notification, payment, TransactionKind.CANCEL
    )
    gateway_postprocess(new_transaction, payment)

    success_msg = "Adyen: The cancel request was successful."
    failed_msg = "Adyen: The request failed."
    create_payment_notification_for_order(
        payment, success_msg, failed_msg, transaction.is_success
    )


def handle_cancel_or_refund(
    notification: Dict[str, Any], gateway_config: GatewayConfig
):
    additional_data = notification.get("additionalData")
    action = additional_data.get("modification.action")
    if action == "refund":
        handle_refund(notification, gateway_config)
    elif action == "cancel":
        handle_cancellation(notification, gateway_config)


def handle_capture(notification: Dict[str, Any], _gateway_config: GatewayConfig):
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    transaction_id = notification.get("pspReference")
    transaction = get_transaction(payment, transaction_id, TransactionKind.CAPTURE)
    if transaction and transaction.is_success:
        # it is already captured
        return

    new_transaction = create_new_transaction(
        notification, payment, TransactionKind.CAPTURE
    )
    gateway_postprocess(new_transaction, payment)

    success_msg = f"Adyen: The capture {transaction_id} request was successful."
    failed_msg = f"Adyen: The capture {transaction_id} request failed."
    create_payment_notification_for_order(
        payment, success_msg, failed_msg, transaction.is_success
    )


def handle_failed_capture(notification: Dict[str, Any], _gateway_config: GatewayConfig):
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    transaction_id = notification.get("pspReference")

    transaction = get_transaction(
        payment, transaction_id, TransactionKind.CAPTURE_FAILED
    )
    if transaction and transaction.is_success:
        # it is already failed
        return
    new_transaction = create_new_transaction(
        notification, payment, TransactionKind.CAPTURE_FAILED
    )
    gateway_postprocess(new_transaction, payment)

    msg = f"Adyen: The capture for {transaction_id} failed due to a technical issue."
    create_payment_notification_for_order(payment, msg, None, True)


def handle_pending(notification: Dict[str, Any], gateway_config: GatewayConfig):
    mark_capture = gateway_config.auto_capture
    if mark_capture:
        # If we mark order as a capture by default we don't need to handle this action
        return
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    transaction_id = notification.get("pspReference")
    transaction = get_transaction(payment, transaction_id, TransactionKind.PENDING)
    if transaction and transaction.is_success:
        # it is already pending
        return
    new_transaction = create_new_transaction(
        notification, payment, TransactionKind.PENDING
    )
    gateway_postprocess(new_transaction, payment)

    msg = f"Adyen: The transaction {transaction_id} is pending."
    create_payment_notification_for_order(payment, msg, None, transaction.is_success)


def handle_refund(notification: Dict[str, Any], _gateway_config: GatewayConfig):
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    transaction_id = notification.get("pspReference")
    transaction = get_transaction(payment, transaction_id, TransactionKind.REFUND)
    if transaction and transaction.is_success:
        # it is already refunded
        return
    new_transaction = create_new_transaction(
        notification, payment, TransactionKind.REFUND
    )
    gateway_postprocess(new_transaction, payment)

    success_msg = f"Adyen: The refund {transaction_id} request was successful."
    failed_msg = f"Adyen: The refund {transaction_id} request failed."
    create_payment_notification_for_order(
        payment, success_msg, failed_msg, transaction.is_success
    )


def handle_failed_refund(notification: Dict[str, Any], _gateway_config: GatewayConfig):
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    transaction_id = notification.get("pspReference")
    transaction = get_transaction(payment, transaction_id, TransactionKind.REFUND)
    if transaction and not transaction.is_success:
        # The refund is already saved
        return
    new_transaction = create_new_transaction(
        notification, payment, TransactionKind.REFUND
    )
    gateway_postprocess(new_transaction, payment)

    msg = (
        f"The refund {transaction_id} failed due to a technical issue. If you receive "
        f"more than two failures on the same refund, contact Adyen Support Team."
    )
    create_payment_notification_for_order(payment, msg, msg, transaction.is_success)


def handle_reversed_refund(
    notification: Dict[str, Any], _gateway_config: GatewayConfig
):
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    transaction_id = notification.get("pspReference")
    transaction = get_transaction(
        payment, transaction_id, TransactionKind.REFUND_REVERSED
    )

    if transaction and not transaction.is_success:
        # it is already refunded
        return
    new_transaction = create_new_transaction(
        notification, payment, TransactionKind.REFUND_REVERSED
    )
    gateway_postprocess(new_transaction, payment)
    msg = (
        f"Adyen: The refunded amount from {transaction_id} has been returned to Adyen, "
        f"and is back in your account. This may happen if the shopper's bank account "
        f"is no longer valid"
    )
    create_payment_notification_for_order(payment, msg, msg, transaction.is_success)


def handle_refund_with_data(
    notification: Dict[str, Any], _gateway_config: GatewayConfig
):

    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    transaction_id = notification.get("pspReference")
    transaction = get_transaction(payment, transaction_id, TransactionKind.REFUND)
    if transaction:
        # it is already refunded
        return
    new_transaction = create_new_transaction(
        notification, payment, TransactionKind.REFUND
    )
    gateway_postprocess(new_transaction, payment)
    success_msg = f"Adyen: The refund {transaction_id} request was successful."
    failed_msg = f"Adyen: The refund {transaction_id} request failed."
    create_payment_notification_for_order(
        payment, success_msg, failed_msg, transaction.is_success
    )


def webhook_not_implemented(
    notification: Dict[str, Any], gateway_config: GatewayConfig
):
    adyen_id = notification.get("pspReference")
    success = notification.get("success", True)
    event = notification.get("eventCode")
    payment = get_payment(notification.get("merchantReference"))
    if not payment:
        return
    msg = (
        f"Received not implemented notification from Adyen. Event name: {event}, "
        f"success: {success}, adyen reference: {adyen_id}."
    )
    create_payment_notification_for_order(payment, msg, None, True)


EVENT_MAP = {
    "AUTHORISATION": handle_authorization,
    "AUTHORISATION_ADJUSTMENT": webhook_not_implemented,
    "CANCELLATION": handle_cancellation,
    "CANCEL_OR_REFUND": handle_cancel_or_refund,
    "CAPTURE": handle_capture,
    "CAPTURE_FAILED": handle_failed_capture,
    "HANDLED_EXTERNALLY": webhook_not_implemented,
    "ORDER_OPENED": webhook_not_implemented,
    "ORDER_CLOSED": webhook_not_implemented,
    "PENDING": handle_pending,
    "PROCESS_RETRY": webhook_not_implemented,
    "REFUND": handle_refund,
    "REFUND_FAILED": handle_failed_refund,
    "REFUNDED_REVERSED": handle_reversed_refund,
    "REFUND_WITH_DATA": handle_refund_with_data,
    "REPORT_AVAILABLE": webhook_not_implemented,
    "VOID_PENDING_REFUND": webhook_not_implemented,
}


def validate_hmac_signature(
    notification: Dict[str, Any], gateway_config: "GatewayConfig"
) -> bool:

    """
    pspReference	7914073381342284
    originalReference
    merchantAccountCode	YOUR_MERCHANT_ACCOUNT
    merchantReference	TestPayment-1407325143704
    value	1130
    currency	EUR
    eventCode	AUTHORISATION
    success	true
    """
    hmac_key = gateway_config.connection_params.get("webhook_hmac")
    if not hmac_key:
        return True

    hmac_signature = notification.get("additionalData", {}).get("hmacSignature")
    if not hmac_signature and hmac_key:
        return False

    success = "true" if notification.get("success", "") else "false"
    if notification.get("success", None) is None:
        success = ""

    payload_list = [
        notification.get("pspReference", ""),
        notification.get("originalReference", ""),
        notification.get("merchantAccountCode", ""),
        notification.get("merchantReference", ""),
        notification.get("value", ""),
        notification.get("currency", ""),
        notification.get("eventCode", ""),
        success,
    ]
    payload = ":".join(payload_list)

    hm = hmac.new(hmac_key, payload.encode("utf-8"), hashlib.sha256)
    expected_merchant_sign = base64.b64encode(hm.digest())
    return hmac_signature == expected_merchant_sign.decode("utf-8")


def validate_auth_user(headers: HttpHeaders, gateway_config: "GatewayConfig") -> bool:
    username = gateway_config.connection_params["webhook_user"]
    password = gateway_config.connection_params["webhook_user_password"]
    auth_header = headers.get("Authorization")
    if not auth_header and not username:
        return True

    split_auth = auth_header.split(maxsplit=1)
    prefix = "BASIC"

    if len(split_auth) != 2 or split_auth[0].upper() != prefix:
        return False

    auth = split_auth[1]
    try:
        request_username, request_password = base64.b64decode(auth).split(":")
        user_is_correct = request_username == username
        if user_is_correct and check_password(request_password, password):
            return True
    except binascii.Error:
        pass
    return False


def handle_webhook(request: WSGIRequest, gateway_config: "GatewayConfig"):
    json_data = json.loads(request.body)
    # JSON and HTTP POST notifications always contain a single NotificationRequestItem
    # object.
    notification = json_data.get("notificationItems")[0].get(
        "NotificationRequestItem", {}
    )

    if not validate_hmac_signature(notification, gateway_config):
        return HttpResponseBadRequest("Invalid or missing hmac signature.")
    if not validate_auth_user(notification, gateway_config):
        return HttpResponseBadRequest("Invalid or missing basic auth.")

    event_handler = EVENT_MAP.get(notification.get("eventCode", ""))
    if event_handler:
        event_handler(notification, gateway_config)
        return HttpResponse("[accepted]")
    return HttpResponseNotFound()
# Generated by Django 3.0.6 on 2020-07-21 13:13

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ("payment", "0018_auto_20200720_1118"),
    ]

    operations = [
        migrations.AlterField(
            model_name="payment",
            name="charge_status",
            field=models.CharField(
                choices=[
                    ("not-charged", "Not charged"),
                    ("pending", "Pending"),
                    ("partially-charged", "Partially charged"),
                    ("fully-charged", "Fully charged"),
                    ("partially-refunded", "Partially refunded"),
                    ("fully-refunded", "Fully refunded"),
                    ("refused", "Refused"),
                    ("cancelled", "Cancelled"),
                ],
                default="not-charged",
                max_length=20,
            ),
        ),
        migrations.AlterField(
            model_name="transaction",
            name="kind",
            field=models.CharField(
                choices=[
                    ("auth", "Authorization"),
                    ("pending", "Pending"),
                    ("refund", "Refund"),
                    ("refund_ongoing", "Refund in progress"),
                    ("capture", "Capture"),
                    ("void", "Void"),
                    ("confirm", "Confirm"),
                    ("cancel", "Cancel"),
                ],
                max_length=25,
            ),
        ),
    ]
    payment_information: Optional[PaymentData],
    if transaction_kind in {
        TransactionKind.CAPTURE,
        TransactionKind.CONFIRM,
        TransactionKind.REFUND_REVERSED,
    }:
    elif transaction_kind == TransactionKind.CANCEL:
        payment.charge_status = ChargeStatus.CANCELLED
        payment.save(
            update_fields=["charge_status",]
        )
    elif transaction_kind == TransactionKind.CAPTURE_FAILED:
        if payment.charge_status in {
            ChargeStatus.PARTIALLY_CHARGED,
            ChargeStatus.FULLY_CHARGED,
        }:
            payment.captured_amount -= transaction.amount
            payment.charge_status = ChargeStatus.PARTIALLY_CHARGED
            if payment.get_charge_amount() <= 0:
                payment.charge_status = ChargeStatus.FULLY_CHARGED

            payment.save(update_fields=["charge_status", "captured_amount", "modified"])