pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[IMP] payment_*: improve handling of webhook notifications

Browse Source 
Notification handling in some acquirers presents a subset of the
following issues:
1. The signature of synchronous notifications (redirect payloads) is not
   checked. (Alipay, Authorize, Buckaroo, Mollie, PayU money, PayULatam)
2. When the signature check fails, we raise a ValidationError which
   counts as an HTTP 200 for some providers (it's not the case if they
   expect a specific string). (Adyen, Paypal,  Sips, Stripe)
3. If a ValidationError is raised when processing the feedback data, it
   is allowed to bubble up to the provider. (Alipay, Ogone)

The issues are respectively addressed as follows:
1. If the acquirer implements payments with redirection, make sure that
   if either makes a request to the provider to validate the data or
   that it verifies the signature. Verifying the origin of the request
   is not enough: the payload must be checked too.
2. Instead of raising ValidationError's, raise an HTTP 403 FORBIDDEN
   error if the signature check fails.
3. Wrap the call to `_handle_feedback_data` of the webhook method inside
   a try/except clause to catch any ValidationError, log a warning, and
   acknowledge the notification to avoid having the provider disable the
   webhook because of too many failures.

task-2688139
task-2693293

closes odoo/odoo#81607

Signed-off-by: Antoine Vandevenne (anv) <anv@odoo.com>
Co-authored-by: Lucie Van Nieuwenhuyze <luvn@odoo.com>
This commit is contained in:
Antoine Vandevenne (anv)
2022-01-27 17:11:52 +00:00
co-authored by Lucie Van Nieuwenhuyze
parent ced94b7863
commit 00259dc44a
43 changed files with 1273 additions and 705 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/payment_adyen/controllers/main.py
+39 -38
View File
@@ -4,17 +4,15 @@ import base64
import binascii
import hashlib
import hmac
import json
import logging
import pprint
import werkzeug
from werkzeug import urls
from werkzeug.exceptions import Forbidden
from odoo import _, http
from odoo.exceptions import ValidationError
from odoo.http import request
from odoo.tools.pycompat import to_text
from odoo.addons.payment import utils as payment_utils
from odoo.addons.payment_adyen.const import CURRENCY_DECIMALS
@@ -24,6 +22,8 @@ _logger = logging.getLogger(__name__)
class AdyenController(http.Controller):
_webhook_url = '/payment/adyen/notification'
@http.route('/payment/adyen/acquirer_info', type='json', auth='public')
def adyen_acquirer_info(self, acquirer_id):
""" Return public information on the acquirer.
@@ -186,8 +186,8 @@ class AdyenController(http.Controller):
return response_content
@http.route('/payment/adyen/return', type='http', auth='public', csrf=False, save_session=False)
def adyen_return_from_redirect(self, **data):
""" Process the data returned by Adyen after redirection.
def adyen_return_from_3ds_auth(self, **data):
""" Process the feedback data sent by Adyen after redirection from 3DS1 authentication page.
The route is flagged with `save_session=False` to prevent Odoo from assigning a new session
to the user if they are redirected to this route with a POST request. Indeed, as the session
@@ -229,7 +229,7 @@ class AdyenController(http.Controller):
# Redirect the user to the status page
return request.redirect('/payment/status')
@http.route('/payment/adyen/notification', type='json', auth='public')
@http.route(_webhook_url, type='json', auth='public')
def adyen_notification(self):
""" Process the data sent by Adyen to the webhook based on the event code.
@@ -239,31 +239,24 @@ class AdyenController(http.Controller):
:return: The '[accepted]' string to acknowledge the notification
:rtype: str
"""
data = json.loads(request.httprequest.data)
data = request.jsonrequest
for notification_item in data['notificationItems']:
notification_data = notification_item['NotificationRequestItem']
# Check the source and integrity of the notification
received_signature = notification_data.get('additionalData', {}).get('hmacSignature')
PaymentTransaction = request.env['payment.transaction']
_logger.info(
"notification received from Adyen with data:\n%s", pprint.pformat(notification_data)
)
try:
# Check the integrity of the notification
PaymentTransaction = request.env['payment.transaction']
tx_sudo = PaymentTransaction.sudo()._get_tx_from_feedback_data(
'adyen', notification_data
)
acquirer_sudo = tx_sudo.acquirer_id # Find the acquirer based on the transaction
if not self._verify_notification_signature(
received_signature, notification_data, acquirer_sudo.adyen_hmac_key
):
continue
self._verify_notification_signature(notification_data, acquirer_sudo.adyen_hmac_key)
# Check whether the event of the notification succeeded and reshape the notification
# data for parsing
_logger.info(
"notification received from Adyen for transaction with reference %s with "
"data:\n%s",
tx_sudo.reference, pprint.pformat(notification_data)
)
success = notification_data['success'] == 'true'
event_code = notification_data['eventCode']
if event_code == 'AUTHORISATION' and success:
@@ -282,18 +275,38 @@ class AdyenController(http.Controller):
return '[accepted]' # Acknowledge the notification
def _verify_notification_signature(self, received_signature, payload, hmac_key):
""" Check that the signature computed from the payload matches the received one.
@staticmethod
def _verify_notification_signature(notification_data, hmac_key):
""" Check that the received signature matches the expected one.
:param dict notification_data: The notification payload containing the received signature
:param str hmac_key: The HMAC key of the acquirer handling the transaction
:return: None
:raise: :class:`werkzeug.exceptions.Forbidden` if the signatures don't match
"""
# Retrieve the received signature from the payload
received_signature = notification_data.get('additionalData', {}).get('hmacSignature')
if not received_signature:
_logger.warning("received notification with missing signature")
raise Forbidden()
# Compare the received signature with the expected signature computed from the payload
expected_signature = AdyenController._compute_signature(notification_data, hmac_key)
if not hmac.compare_digest(received_signature, expected_signature):
_logger.warning("received notification with invalid signature")
raise Forbidden()
@staticmethod
def _compute_signature(payload, hmac_key):
""" Compute the signature from the payload.
See https://docs.adyen.com/development-resources/webhooks/verify-hmac-signatures
:param str received_signature: The signature sent with the notification
:param dict payload: The notification payload
:param str hmac_key: The HMAC key of the acquirer handling the transaction
:return: Whether the signatures match
:return: The computed signature
:rtype: str
"""
def _flatten_dict(_value, _path_base='', _separator='.'):
""" Recursively generate a flat representation of a dict.
@@ -325,11 +338,6 @@ class AdyenController(http.Controller):
else:
return str(_value)
if not received_signature:
_logger.warning("ignored notification with missing signature")
return False
# Compute the signature from the payload
signature_keys = [
'pspReference', 'originalReference', 'merchantAccountCode', 'merchantReference',
'amount.value', 'amount.currency', 'eventCode', 'success'
@@ -347,11 +355,4 @@ class AdyenController(http.Controller):
# Calculate the HMAC with the binary representation of the signing string with SHA-256
binary_hmac = hmac.new(binary_hmac_key, signing_string.encode('utf-8'), hashlib.sha256)
# Calculate the signature by encoding the result with Base64
expected_signature = base64.b64encode(binary_hmac.digest())
# Compare signatures
if received_signature != to_text(expected_signature):
_logger.warning("ignored event with invalid signature")
return False
return True
return base64.b64encode(binary_hmac.digest()).decode()
addons/payment_adyen/tests/common.py
+25 -1
View File
@@ -1,9 +1,33 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.addons.payment.tests.common import PaymentCommon
class AdyenCommon(PaymentCommon):
WEBHOOK_NOTIFICATION_PAYLOAD = {
'additionalData': {
'hmacSignature': 'kK6vSQvfWP3AtT2TTK1ePj9e7XPb7bF5jHC7jDWyU5c='
},
'amount': {
'currency': 'USD',
'value': 999,
},
'eventCode': 'AUTHORISATION',
'merchantAccountCode': 'DuckSACom123',
'merchantReference': 'Test Transaction', # Shamefully copy-pasted from payment
'originalReference': 'FEDCBA9876543210',
'pspReference': '0123456789ABCDEF',
'success': 'true',
} # Include all keys used in the computation of the signature to the payload
WEBHOOK_NOTIFICATION_BATCH_DATA = {
'notificationItems': [
{
'NotificationRequestItem': WEBHOOK_NOTIFICATION_PAYLOAD,
}
]
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
@@ -12,7 +36,7 @@ class AdyenCommon(PaymentCommon):
'adyen_merchant_account': 'dummy',
'adyen_api_key': 'dummy',
'adyen_client_key': 'dummy',
'adyen_hmac_key': 'dummy',
'adyen_hmac_key': '12345678',
'adyen_checkout_api_url': 'https://this.is.an.url',
'adyen_recurring_api_url': 'https://this.is.an.url',
})
addons/payment_adyen/tests/test_adyen.py
+64 -3
View File
@@ -2,17 +2,20 @@
from unittest.mock import patch
from werkzeug.exceptions import Forbidden
from odoo.exceptions import UserError
from odoo.tests import tagged
from odoo.tools import mute_logger
from odoo.addons.payment import utils as payment_utils
from .common import AdyenCommon
from odoo.addons.payment.tests.http_common import PaymentHttpCommon
from odoo.addons.payment_adyen.controllers.main import AdyenController
from odoo.addons.payment_adyen.tests.common import AdyenCommon
@tagged('post_install', '-at_install')
class AdyenForm(AdyenCommon):
class AdyenTest(AdyenCommon, PaymentHttpCommon):
def test_processing_values(self):
tx = self.create_transaction(flow='direct')
@@ -68,3 +71,61 @@ class AdyenForm(AdyenCommon):
msg="The acquirer reference of the refund transaction should different from that of "
"the source transaction."
)
@mute_logger('odoo.addons.payment_adyen.controllers.main')
def test_webhook_notification_confirms_transaction(self):
""" Test the processing of a webhook notification. """
tx = self.create_transaction('direct')
url = self._build_url(AdyenController._webhook_url)
with patch(
'odoo.addons.payment_adyen.controllers.main.AdyenController'
'._verify_notification_signature'
):
self._make_json_request(url, data=self.WEBHOOK_NOTIFICATION_BATCH_DATA)
self.assertEqual(tx.state, 'done')
@mute_logger('odoo.addons.payment_adyen.controllers.main')
def test_webhook_notification_triggers_signature_check(self):
""" Test that receiving a webhook notification triggers a signature check. """
self.create_transaction('direct')
url = self._build_url(AdyenController._webhook_url)
with patch(
'odoo.addons.payment_adyen.controllers.main.AdyenController'
'._verify_notification_signature'
) as signature_check_mock, patch(
'odoo.addons.payment.models.payment_transaction.PaymentTransaction'
'._handle_feedback_data'
):
self._make_json_request(url, data=self.WEBHOOK_NOTIFICATION_BATCH_DATA)
self.assertEqual(signature_check_mock.call_count, 1)
def test_accept_webhook_notification_with_valid_signature(self):
""" Test the verification of a webhook notification with a valid signature. """
self._assert_does_not_raise(
Forbidden,
AdyenController._verify_notification_signature,
self.WEBHOOK_NOTIFICATION_PAYLOAD,
self.acquirer.adyen_hmac_key,
)
@mute_logger('odoo.addons.payment_adyen.controllers.main')
def test_reject_webhook_notification_with_missing_signature(self):
""" Test the verification of a webhook notification with a missing signature. """
payload = dict(self.WEBHOOK_NOTIFICATION_PAYLOAD, additionalData={'hmacSignature': None})
self.assertRaises(
Forbidden,
AdyenController._verify_notification_signature,
payload,
self.acquirer.adyen_hmac_key,
)
@mute_logger('odoo.addons.payment_adyen.controllers.main')
def test_reject_webhook_notification_with_invalid_signature(self):
""" Test the verification of a webhook notification with an invalid signature. """
payload = dict(self.WEBHOOK_NOTIFICATION_PAYLOAD, additionalData={'hmacSignature': 'dummy'})
self.assertRaises(
Forbidden,
AdyenController._verify_notification_signature,
payload,
self.acquirer.adyen_hmac_key,
)
addons/payment_alipay/controllers/main.py
+96 -35
View File
@@ -1,11 +1,13 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import hmac
import logging
import pprint
import requests
from werkzeug.exceptions import Forbidden
from odoo import _, http
from odoo import http
from odoo.exceptions import ValidationError
from odoo.http import request
@@ -14,48 +16,107 @@ _logger = logging.getLogger(__name__)
class AlipayController(http.Controller):
_return_url = '/payment/alipay/return'
_notify_url = '/payment/alipay/notify'
_webhook_url = '/payment/alipay/webhook'
@http.route(_return_url, type='http', auth="public", methods=['GET'])
def alipay_return_from_redirect(self, **data):
""" Alipay return """
@http.route(_return_url, type='http', auth='public', methods=['GET'])
def alipay_return_from_checkout(self, **data):
""" Process the notification data sent by Alipay after redirection from checkout.
See https://global.alipay.com/docs/ac/web/sync.
:param dict data: The notification data
"""
_logger.info("handling redirection from Alipay with data:\n%s", pprint.pformat(data))
# Check the integrity of the notification
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'alipay', data
)
self._verify_notification_signature(data, tx_sudo)
# Handle the notification data
request.env['payment.transaction'].sudo()._handle_feedback_data('alipay', data)
return request.redirect('/payment/status')
@http.route(_notify_url, type='http', auth='public', methods=['POST'], csrf=False)
def alipay_notify(self, **post):
""" Alipay Notify """
_logger.info("notification received from Alipay with data:\n%s", pprint.pformat(post))
self._alipay_validate_notification(**post)
request.env['payment.transaction'].sudo()._handle_feedback_data('alipay', post)
return 'success' # Return 'success' to stop receiving notifications for this tx
@http.route(_webhook_url, type='http', auth='public', methods=['POST'], csrf=False)
def alipay_webhook(self, **data):
""" Process the notification data sent by Alipay to the webhook.
def _alipay_validate_notification(self, **post):
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'alipay', post
)
if not tx_sudo:
raise ValidationError(
"Alipay: " + _(
"Received notification data with unknown reference:\n%s", pprint.pformat(post)
)
See https://global.alipay.com/docs/ac/web/async.
:param dict data: The notification data
:return: The 'SUCCESS' string to acknowledge the notification
:rtype: str
"""
_logger.info("notification received from Alipay with data:\n%s", pprint.pformat(data))
try:
# Check the origin and integrity of the notification
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'alipay', data
)
self._verify_notification_origin(data, tx_sudo)
self._verify_notification_signature(data, tx_sudo)
# Ensure that the notification was sent by Alipay
# See https://global.alipay.com/docs/ac/wap/async
acquirer_sudo = tx_sudo.acquirer_id
val = {
# Handle the notification data
request.env['payment.transaction'].sudo()._handle_feedback_data('alipay', data)
except ValidationError: # Acknowledge the notification to avoid getting spammed
_logger.exception("unable to handle the notification data; skipping to acknowledge")
return 'SUCCESS' # Acknowledge the notification
@staticmethod
def _verify_notification_origin(notification_data, tx_sudo):
""" Check that the notification was sent by Alipay.
See https://global.alipay.com/docs/ac/web/async#9727f6bd.
:param dict notification_data: The notification data
:param recordset tx_sudo: The sudoed transaction referenced in the notification data, as a
`payment.transaction` record
:return: None
:raise: :class:`werkzeug.exceptions.Forbidden` if the notification origin can't be verified
"""
url = tx_sudo.acquirer_id._alipay_get_api_url()
payload = {
'service': 'notify_verify',
'partner': acquirer_sudo.alipay_merchant_partner_id,
'notify_id': post['notify_id']
'partner': tx_sudo.acquirer_id.alipay_merchant_partner_id,
'notify_id': notification_data['notify_id'],
}
response = requests.post(acquirer_sudo._alipay_get_api_url(), val, timeout=60)
response.raise_for_status()
if response.text != 'true':
raise ValidationError(
"Alipay: " + _(
"Received notification data not acknowledged by Alipay:\n%s",
pprint.pformat(post)
)
try:
response = requests.post(url, data=payload, timeout=60)
response.raise_for_status()
except (requests.exceptions.ConnectionError, requests.exceptions.HTTPError) as error:
_logger.exception(
"could not verify notification origin at %(url)s with data: %(data)s:\n%(error)s",
{'url': url, 'data': payload, 'error': pprint.pformat(error.response.text)},
)
raise Forbidden()
else:
response_content = response.text
if response_content != 'true':
_logger.error(
"Alipay did not confirm the origin of the notification with data:\n%s", payload
)
raise Forbidden()
@staticmethod
def _verify_notification_signature(notification_data, tx_sudo):
""" Check that the received signature matches the expected one.
:param dict notification_data: The notification data
:param recordset tx_sudo: The sudoed transaction referenced by the notification data, as a
`payment.transaction` record
:return: None
:raise: :class:`werkzeug.exceptions.Forbidden` if the signatures don't match
"""
# Retrieve the received signature from the data
received_signature = notification_data.get('sign')
if not received_signature:
_logger.warning("received notification with missing signature")
raise Forbidden()
# Compare the received signature with the expected signature computed from the data
expected_signature = tx_sudo.acquirer_id._alipay_compute_signature(notification_data)
if not hmac.compare_digest(received_signature, expected_signature):
_logger.warning("received notification with invalid signature")
raise Forbidden()
addons/payment_alipay/models/payment_acquirer.py
+2 -2
View File
@@ -42,9 +42,9 @@ class PaymentAcquirer(models.Model):
return acquirers
def _alipay_build_sign(self, val):
def _alipay_compute_signature(self, data):
# Rearrange parameters in the data set alphabetically
data_to_sign = sorted(val.items())
data_to_sign = sorted(data.items())
# Format key-value pairs of parameters that should be signed
data_to_sign = [f"{k}={v}" for k, v in data_to_sign
if k not in ['sign', 'sign_type', 'reference']]
addons/payment_alipay/models/payment_transaction.py
+2 -47
View File
@@ -39,7 +39,7 @@ class PaymentTransaction(models.Model):
total_fee = self.amount
rendering_values = {
'_input_charset': 'utf-8',
'notify_url': urls.url_join(base_url, AlipayController._notify_url),
'notify_url': urls.url_join(base_url, AlipayController._webhook_url),
'out_trade_no': self.reference,
'partner': self.acquirer_id.alipay_merchant_partner_id,
'return_url': urls.url_join(base_url, AlipayController._return_url),
@@ -60,7 +60,7 @@ class PaymentTransaction(models.Model):
'seller_email': self.acquirer_id.alipay_seller_email,
})
sign = self.acquirer_id._alipay_build_sign(rendering_values)
sign = self.acquirer_id._alipay_compute_signature(rendering_values)
rendering_values.update({
'sign_type': 'MD5',
'sign': sign,
@@ -99,16 +99,6 @@ class PaymentTransaction(models.Model):
"Alipay: " + _("No transaction found matching reference %s.", reference)
)
# Verify signature (done here because we need the reference to get the acquirer)
sign_check = tx.acquirer_id._alipay_build_sign(data)
sign = data.get('sign')
if sign != sign_check:
raise ValidationError(
"Alipay: " + _(
"Expected signature %(sc) but received %(sign)s.", sc=sign_check, sign=sign
)
)
return tx
def _process_feedback_data(self, data):
@@ -124,41 +114,6 @@ class PaymentTransaction(models.Model):
if self.provider != 'alipay':
return
if float_compare(float(data.get('total_fee', '0.0')), (self.amount + self.fees), 2) != 0:
# mc_gross is amount + fees
_logger.error(
"the paid amount (%(amount)s) does not match the total + fees (%(total)s + "
"%(fees)s) for transaction with reference %(ref)s",
{
'amount': data.get('total_fee', '0.0'),
'total': self.amount,
'fees': self.fees,
'ref': self.reference,
}
)
raise ValidationError("Alipay: " + _("The amount does not match the total + fees."))
if self.acquirer_id.alipay_payment_method == 'standard_checkout':
if data.get('currency') != self.currency_id.name:
raise ValidationError(
"Alipay: " + _(
"The currency returned by Alipay %(rc)s does not match the transaction "
"currency %(tc)s.", rc=data.get('currency'), tc=self.currency_id.name
)
)
elif data.get('seller_email') != self.acquirer_id.alipay_seller_email:
_logger.error(
"the seller email (%(email)s) does not match the configured Alipay account "
"(%(acc_email)s) for transaction with reference %(ref)s",
{
'email': data.get('seller_email'),
'acc_email:': self.acquirer_id.alipay_seller_email,
'ref': self.reference,
},
)
raise ValidationError(
"Alipay: " + _("The seller email does not match the configured Alipay account.")
)
self.acquirer_reference = data.get('trade_no')
status = data.get('trade_status')
if status in ['TRADE_FINISHED', 'TRADE_SUCCESS']:
addons/payment_alipay/tests/common.py
+14 -1
View File
@@ -5,6 +5,19 @@ from odoo.addons.payment.tests.common import PaymentCommon
class AlipayCommon(PaymentCommon):
NOTIFICATION_DATA = {
'currency': 'CNY',
'notify_id': '1234567890123456789012345678901234',
'notify_time': '2021-12-01 01:01:01',
'notify_type': 'trade_status_sync',
'out_trade_no': 'Test Transaction', # Shamefully copy-pasted from payment
'sign': '782b6d1015549f847e2ab27d1edb65c7',
'sign_type': 'MD5',
'total_fee': '1111.11',
'trade_no': '2021111111111111111111111111',
'trade_status': 'TRADE_FINISHED',
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
@@ -14,7 +27,7 @@ class AlipayCommon(PaymentCommon):
'alipay_merchant_partner_id': 'dummy',
'alipay_md5_signature_key': 'dummy',
'alipay_seller_email': 'dummy',
'fees_active': False, # Only activate fees in dedicated tests
'fees_active': False, # Only activate fees in dedicated tests
})
# override defaults for helpers
addons/payment_alipay/tests/test_alipay.py
+82 -59
View File
@@ -1,21 +1,26 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from unittest.mock import patch
from werkzeug.exceptions import Forbidden
from odoo.exceptions import ValidationError
from odoo.tests import tagged
from odoo.tools import mute_logger
from .common import AlipayCommon
from ..controllers.main import AlipayController
from odoo.addons.payment.tests.http_common import PaymentHttpCommon
from odoo.addons.payment_alipay.controllers.main import AlipayController
from odoo.addons.payment_alipay.tests.common import AlipayCommon
@tagged('post_install', '-at_install')
class AlipayTest(AlipayCommon):
class AlipayTest(AlipayCommon, PaymentHttpCommon):
def test_compatible_acquirers(self):
self.alipay.alipay_payment_method = 'express_checkout'
acquirers = self.env['payment.acquirer']._get_compatible_acquirers(
partner_id=self.partner.id,
currency_id=self.currency_yuan.id, # 'CNY'
currency_id=self.currency_yuan.id, # 'CNY'
company_id=self.company.id,
)
self.assertIn(self.alipay, acquirers)
@@ -29,7 +34,7 @@ class AlipayTest(AlipayCommon):
self.alipay.alipay_payment_method = 'standard_checkout'
acquirers = self.env['payment.acquirer']._get_compatible_acquirers(
partner_id=self.partner.id,
currency_id=self.currency_yuan.id, # 'CNY'
currency_id=self.currency_yuan.id, # 'CNY'
company_id=self.company.id,
)
self.assertIn(self.alipay, acquirers)
@@ -49,16 +54,16 @@ class AlipayTest(AlipayCommon):
self._test_alipay_redirect_form()
def _test_alipay_redirect_form(self):
tx = self.create_transaction(flow='redirect') # Only flow implemented
tx = self.create_transaction(flow='redirect') # Only flow implemented
expected_values = {
'_input_charset': 'utf-8',
'notify_url': self._build_url(AlipayController._notify_url),
'notify_url': self._build_url(AlipayController._webhook_url),
'out_trade_no': self.reference,
'partner': self.alipay.alipay_merchant_partner_id,
'return_url': self._build_url(AlipayController._return_url),
'subject': self.reference,
'total_fee': str(self.amount), # Fees disabled by default
'total_fee': str(self.amount), # Fees disabled by default
}
if self.alipay.alipay_payment_method == 'standard_checkout':
@@ -73,7 +78,7 @@ class AlipayTest(AlipayCommon):
'payment_type': str(1),
'seller_email': self.alipay.alipay_seller_email,
})
sign = self.alipay._alipay_build_sign(expected_values)
sign = self.alipay._alipay_compute_signature(expected_values)
with mute_logger('odoo.addons.payment.models.payment_transaction'):
processing_values = tx._get_processing_values()
@@ -134,60 +139,78 @@ class AlipayTest(AlipayCommon):
self._test_alipay_feedback_processing()
def _test_alipay_feedback_processing(self):
# typical data posted by alipay after client has successfully paid
custom_reference = 'test_ref_' + self.alipay.alipay_payment_method
alipay_post_data = {
'trade_no': '2017112321001003690200384552',
'reference': custom_reference,
'total_fee': 1.95,
'trade_status': 'TRADE_CLOSED',
}
# Unknown transaction
with self.assertRaises(ValidationError):
self.env['payment.transaction']._handle_feedback_data('alipay', self.NOTIFICATION_DATA)
if self.alipay.alipay_payment_method == 'express_checkout':
alipay_post_data.update({
'seller_email': self.alipay.alipay_seller_email,
})
else:
alipay_post_data.update({
'currency': 'EUR',
})
# Confirmed transaction
tx = self.create_transaction('redirect')
self.env['payment.transaction']._handle_feedback_data('alipay', self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
self.assertEqual(tx.acquirer_reference, self.NOTIFICATION_DATA['trade_no'])
alipay_post_data['sign'] = self.alipay._alipay_build_sign(alipay_post_data)
with self.assertRaises(ValidationError): # unknown transactiion
self.env['payment.transaction']._handle_feedback_data('alipay', alipay_post_data)
# Pending transaction
self.reference = 'Test Transaction 2'
tx = self.create_transaction('redirect')
payload = dict(
self.NOTIFICATION_DATA, out_trade_no=self.reference, trade_status='TRADE_CLOSED'
)
self.env['payment.transaction']._handle_feedback_data('alipay', payload)
self.assertEqual(tx.state, 'cancel')
tx = self.env['payment.transaction'].create({
'amount': 1.95,
'acquirer_id': self.alipay.id,
'currency_id': self.currency.id,
'reference': custom_reference,
'partner_id': self.partner.id
})
@mute_logger('odoo.addons.payment_alipay.controllers.main')
def test_webhook_notification_confirms_transaction(self):
""" Test the processing of a webhook notification. """
self.acquirer.alipay_payment_method = 'standard_checkout'
tx = self.create_transaction('redirect')
url = self._build_url(AlipayController._webhook_url)
with patch(
'odoo.addons.payment_alipay.controllers.main.AlipayController'
'._verify_notification_origin'
), patch(
'odoo.addons.payment_alipay.controllers.main.AlipayController'
'._verify_notification_signature'
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
self.env['payment.transaction']._handle_feedback_data('alipay', alipay_post_data)
self.assertEqual(tx.state, 'cancel',
'Alipay: wrong state after receiving a valid pending notification')
self.assertEqual(tx.acquirer_reference, '2017112321001003690200384552',
'Alipay: wrong txn_id after receiving a valid pending notification')
@mute_logger('odoo.addons.payment_alipay.controllers.main')
def test_webhook_notification_triggers_origin_and_signature_checks(self):
""" Test that receiving a webhook notification triggers origin and signature checks. """
self.acquirer.alipay_payment_method = 'standard_checkout'
self.create_transaction('redirect')
url = self._build_url(AlipayController._webhook_url)
with patch(
'odoo.addons.payment_alipay.controllers.main.AlipayController'
'._verify_notification_origin'
) as origin_check_mock, patch(
'odoo.addons.payment_alipay.controllers.main.AlipayController'
'._verify_notification_signature'
) as signature_check_mock, patch(
'odoo.addons.payment.models.payment_transaction.PaymentTransaction'
'._handle_feedback_data'
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(origin_check_mock.call_count, 1)
self.assertEqual(signature_check_mock.call_count, 1)
# reset the transaction
tx.write({'state': 'draft', 'acquirer_reference': False})
def test_accept_notification_with_valid_signature(self):
""" Test the verification of a notification with a valid signature. """
tx = self.create_transaction('redirect')
self._assert_does_not_raise(
Forbidden, AlipayController._verify_notification_signature, self.NOTIFICATION_DATA, tx
)
# update notification from alipay should not go through since it has already been set as 'done'
if self.alipay.alipay_payment_method == 'standard_checkout':
alipay_post_data['trade_status'] = 'TRADE_FINISHED'
else:
alipay_post_data['trade_status'] = 'TRADE_SUCCESS'
alipay_post_data['sign'] = self.alipay._alipay_build_sign(alipay_post_data)
@mute_logger('odoo.addons.payment_alipay.controllers.main')
def test_reject_notification_with_missing_signature(self):
""" Test the verification of a notification with a missing signature. """
tx = self.create_transaction('redirect')
payload = dict(self.NOTIFICATION_DATA, sign=None)
self.assertRaises(Forbidden, AlipayController._verify_notification_signature, payload, tx)
self.env['payment.transaction']._handle_feedback_data('alipay', alipay_post_data)
self.assertEqual(tx.acquirer_reference, '2017112321001003690200384552',
'Alipay: notification should not go throught since it has already been validated')
# this time it should go through since the transaction is not validated yet
tx.write({'state': 'draft', 'acquirer_reference': False})
self.env['payment.transaction']._handle_feedback_data('alipay', alipay_post_data)
self.assertEqual(tx.state, 'done',
'Alipay: wrong state after receiving a valid pending notification')
self.assertEqual(tx.acquirer_reference, '2017112321001003690200384552',
'Alipay: wrong txn_id after receiving a valid pending notification')
@mute_logger('odoo.addons.payment_alipay.controllers.main')
def test_reject_notification_with_invalid_signature(self):
""" Test the verification of a notification with an invalid signature. """
tx = self.create_transaction('redirect')
payload = dict(self.NOTIFICATION_DATA, sign='dummy')
self.assertRaises(Forbidden, AlipayController._verify_notification_signature, payload, tx)
addons/payment_buckaroo/controllers/main.py
+54 -4
View File
@@ -1,8 +1,11 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import hmac
import logging
import pprint
from werkzeug.exceptions import Forbidden
from odoo import http
from odoo.http import request
@@ -15,8 +18,8 @@ class BuckarooController(http.Controller):
@http.route(
_return_url, type='http', auth='public', methods=['POST'], csrf=False, save_session=False
)
def buckaroo_return_from_redirect(self, **data):
""" Process the data returned by Buckaroo after redirection.
def buckaroo_return_from_checkout(self, **raw_data):
""" Process the notification data returned by Buckaroo after redirection from checkout.
The route is flagged with `save_session=False` to prevent Odoo from assigning a new session
to the user if they are redirected to this route with a POST request. Indeed, as the session
@@ -26,8 +29,55 @@ class BuckarooController(http.Controller):
will satisfy any specification of the `SameSite` attribute, the session of the user will be
retrieved and with it the transaction which will be immediately post-processed.
:param dict data: The feedback data
:param dict raw_data: The un-formatted notification data
"""
_logger.info("handling redirection from Buckaroo with data:\n%s", pprint.pformat(data))
_logger.info("handling redirection from Buckaroo with data:\n%s", pprint.pformat(raw_data))
data = self._normalize_data_keys(raw_data)
# Check the integrity of the notification
received_signature = data.get('brq_signature')
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'buckaroo', data
)
self._verify_notification_signature(raw_data, received_signature, tx_sudo)
# Handle the notification data
request.env['payment.transaction'].sudo()._handle_feedback_data('buckaroo', data)
return request.redirect('/payment/status')
@staticmethod
def _normalize_data_keys(data):
""" Set all keys of a dictionary to lower-case.
As Buckaroo parameters names are case insensitive, we can convert everything to lower-case
to easily detected the presence of a parameter by checking the lower-case key only.
:param dict data: The dictionary whose keys must be set to lower-case
:return: A copy of the original data with all keys set to lower-case
:rtype: dict
"""
return {key.lower(): val for key, val in data.items()}
@staticmethod
def _verify_notification_signature(notification_data, received_signature, tx_sudo):
""" Check that the received signature matches the expected one.
:param dict notification_data: The notification data
:param str received_signature: The signature received with the notification data
:param recordset tx_sudo: The sudoed transaction referenced by the notification data, as a
`payment.transaction` record
:return: None
:raise: :class:`werkzeug.exceptions.Forbidden` if the signatures don't match
"""
# Check for the received signature
if not received_signature:
_logger.warning("received notification with missing signature")
raise Forbidden()
# Compare the received signature with the expected signature computed from the data
expected_signature = tx_sudo.acquirer_id._buckaroo_generate_digital_sign(
notification_data, incoming=True
)
if not hmac.compare_digest(received_signature, expected_signature):
_logger.warning("received notification with invalid signature")
raise Forbidden()
addons/payment_buckaroo/models/payment_acquirer.py
+14 -14
View File
@@ -36,27 +36,27 @@ class PaymentAcquirer(models.Model):
""" Generate the shasign for incoming or outgoing communications.
:param dict values: The values used to generate the signature
:param bool incoming: Whether the signature must be generate for an incoming (Buckaroo to
:param bool incoming: Whether the signature must be generated for an incoming (Buckaroo to
Odoo) or outgoing (Odoo to Buckaroo) communication.
:return: The shasign
:rtype: str
"""
if incoming:
# Remove the signature from the values used to check the signature
for key in values.keys():
if key.upper() == 'BRQ_SIGNATURE': # Keys are case-insensitive
del values[key]
break
# Incoming communication values must be URL-decoded before checking the signature
items = [(k, urls.url_unquote_plus(v)) for k, v in values.items()]
else:
# Only use items whose key starts with 'add_', 'brq_', or 'cust_' (case insensitive)
# Incoming communication values must be URL-decoded before checking the signature. The
# key 'brq_signature' must be ignored.
items = [
(k, v) for k, v in values.items()
if any(k.upper().startswith(key_prefix) for key_prefix in ('ADD_', 'BRQ_', 'CUST_'))
(k, urls.url_unquote_plus(v)) for k, v in values.items()
if k.lower() != 'brq_signature'
]
# Sort parameters by lower-cased key. Not upper- because ord('A') < ord('_') < ord('a').
sorted_items = sorted(items, key=lambda pair: pair[0].lower())
else:
items = values.items()
# Only use items whose key starts with 'add_', 'brq_', or 'cust_' (case insensitive)
filtered_items = [
(k, v) for k, v in items
if any(k.lower().startswith(key_prefix) for key_prefix in ('add_', 'brq_', 'cust_'))
]
# Sort parameters by lower-cased key. Not upper-case because ord('A') < ord('_') < ord('a').
sorted_items = sorted(filtered_items, key=lambda pair: pair[0].lower())
# Build the signing string by concatenating all parameters
sign_string = ''.join(f'{k}={v or ""}' for k, v in sorted_items)
# Append the pre-shared secret key to the signing string
addons/payment_buckaroo/models/payment_transaction.py
+9 -42
View File
@@ -13,19 +13,6 @@ from odoo.addons.payment_buckaroo.controllers.main import BuckarooController
_logger = logging.getLogger(__name__)
def _normalize_dataset(data):
""" Set all keys of a dictionary to uppercase.
As Buckaroo parameters names are case insensitive, we can convert everything to upper case to
easily detected the presence of a parameter by checking the uppercase key only.
:param dict data: The dictionary whose keys must be set to uppercase
:return: A copy of the original data with all keys set to uppercase
:rtype: dict
"""
return {key.upper(): val for key, val in data.items()}
class PaymentTransaction(models.Model):
_inherit = 'payment.transaction'
@@ -67,44 +54,22 @@ class PaymentTransaction(models.Model):
""" Override of payment to find the transaction based on Buckaroo data.
:param str provider: The provider of the acquirer that handled the transaction
:param dict data: The feedback data sent by the provider
:param dict data: The normalized feedback data sent by the provider
:return: The transaction if found
:rtype: recordset of `payment.transaction`
:raise: ValidationError if inconsistent data were received
:raise: ValidationError if the data match no transaction
:raise: ValidationError if the signature can not be verified
"""
tx = super()._get_tx_from_feedback_data(provider, data)
if provider != 'buckaroo':
return tx
normalized_data = _normalize_dataset(data)
reference = normalized_data.get('BRQ_INVOICENUMBER')
shasign = normalized_data.get('BRQ_SIGNATURE')
if not reference or not shasign:
raise ValidationError(
"Buckaroo: " + _(
"Received data with missing reference (%(ref)s) or shasign (%(sign)s)",
ref=reference, sign=shasign
)
)
reference = data.get('brq_invoicenumber')
tx = self.search([('reference', '=', reference), ('provider', '=', 'buckaroo')])
if not tx:
raise ValidationError(
"Buckaroo: " + _("No transaction found matching reference %s.", reference)
)
# Verify signature
shasign_check = tx.acquirer_id._buckaroo_generate_digital_sign(data, incoming=True)
if shasign_check != shasign:
raise ValidationError(
"Buckaroo: " + _(
"Invalid shasign: received %(sign)s, computed %(check)s",
sign=shasign, check=shasign_check
)
)
return tx
def _process_feedback_data(self, data):
@@ -112,7 +77,7 @@ class PaymentTransaction(models.Model):
Note: self.ensure_one()
:param dict data: The feedback data sent by the provider
:param dict data: The normalized feedback data sent by the provider
:return: None
:raise: ValidationError if inconsistent data were received
"""
@@ -120,15 +85,14 @@ class PaymentTransaction(models.Model):
if self.provider != 'buckaroo':
return
normalized_data = _normalize_dataset(data)
transaction_keys = normalized_data.get('BRQ_TRANSACTIONS')
transaction_keys = data.get('brq_transactions')
if not transaction_keys:
raise ValidationError("Buckaroo: " + _("Received data with missing transaction keys"))
# BRQ_TRANSACTIONS can hold multiple, comma-separated, tx keys. In practice, it holds only
# one reference. So we split for semantic correctness and keep the first transaction key.
self.acquirer_reference = transaction_keys.split(',')[0]
status_code = int(normalized_data.get('BRQ_STATUSCODE') or 0)
status_code = int(data.get('brq_statuscode') or 0)
if status_code in STATUS_CODES_MAPPING['pending']:
self._set_pending()
elif status_code in STATUS_CODES_MAPPING['done']:
@@ -138,7 +102,10 @@ class PaymentTransaction(models.Model):
elif status_code in STATUS_CODES_MAPPING['refused']:
self._set_error(_("Your payment was refused (code %s). Please try again.", status_code))
elif status_code in STATUS_CODES_MAPPING['error']:
self._set_error(_("An error occurred during processing of your payment (code %s). Please try again.", status_code))
self._set_error(_(
"An error occurred during processing of your payment (code %s). Please try again.",
status_code,
))
else:
_logger.warning(
"received data with invalid payment status (%s) for transaction with reference %s",
addons/payment_buckaroo/tests/common.py
+13
View File
@@ -5,6 +5,19 @@ from odoo.addons.payment.tests.common import PaymentCommon
class BuckarooCommon(PaymentCommon):
SYNC_NOTIFICATION_DATA = {
'brq_payment': 'ABCDEF0123456789ABCDEF0123456789',
'brq_payment_method': 'paypal',
'brq_statuscode': '190', # confirmed
'brq_statusmessage': 'Transaction successfully processed',
'brq_invoicenumber': 'Test Transaction', # Shamefully copy-pasted from payment
'brq_amount': '1111.11',
'brq_currency': 'USD',
'brq_timestamp': '2022-01-01 12:00:00',
'brq_transactions': '0123456789ABCDEF0123456789ABCDEF',
'brq_signature': '5d389aa4f563cd99666a2e6bef79da3d4a32eb50',
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
addons/payment_buckaroo/tests/test_buckaroo.py
+52 -54
View File
@@ -1,11 +1,12 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.exceptions import ValidationError
from werkzeug.exceptions import Forbidden
from odoo.tests import tagged
from odoo.tools import mute_logger
from .common import BuckarooCommon
from ..controllers.main import BuckarooController
from odoo.addons.payment_buckaroo.controllers.main import BuckarooController
from odoo.addons.payment_buckaroo.tests.common import BuckarooCommon
@tagged('post_install', '-at_install')
@@ -35,65 +36,62 @@ class BuckarooTest(BuckarooCommon):
self.assertDictEqual(expected_values, form_info['inputs'],
"Buckaroo: invalid inputs specified in the redirect form.")
@mute_logger('odoo.addons.payment_buckaroo.models.payment_transaction')
def test_feedback_processing(self):
self.amount = 2240.0
self.reference = 'SO004'
# typical data posted by buckaroo after client has successfully paid
buckaroo_post_data = {
'BRQ_RETURNDATA': u'',
'BRQ_AMOUNT': str(self.amount),
'BRQ_CURRENCY': self.currency.name,
'BRQ_CUSTOMER_NAME': u'Jan de Tester',
'BRQ_INVOICENUMBER': self.reference,
'brq_payment': u'573311D081B04069BD6336001611DBD4',
'BRQ_PAYMENT_METHOD': u'paypal',
'BRQ_SERVICE_PAYPAL_PAYERCOUNTRY': u'NL',
'BRQ_SERVICE_PAYPAL_PAYEREMAIL': u'fhe@odoo.com',
'BRQ_SERVICE_PAYPAL_PAYERFIRSTNAME': u'Jan',
'BRQ_SERVICE_PAYPAL_PAYERLASTNAME': u'Tester',
'BRQ_SERVICE_PAYPAL_PAYERMIDDLENAME': u'de',
'BRQ_SERVICE_PAYPAL_PAYERSTATUS': u'verified',
'Brq_signature': u'e67e32ee1be1030a86c7764adfcc01856e00f9a7',
'BRQ_STATUSCODE': u'190',
'BRQ_STATUSCODE_DETAIL': u'S001',
'BRQ_STATUSMESSAGE': u'Transaction successfully processed',
'BRQ_TIMESTAMP': u'2014-05-08 12:41:21',
'BRQ_TRANSACTIONS': u'D6106678E1D54EEB8093F5B3AC42EA7B',
'BRQ_WEBSITEKEY': u'5xTGyGyPyl',
}
with self.assertRaises(ValidationError): # unknown transaction
self.env['payment.transaction']._handle_feedback_data('buckaroo', buckaroo_post_data)
notification_data = BuckarooController._normalize_data_keys(self.SYNC_NOTIFICATION_DATA)
tx = self.create_transaction(flow='redirect')
# validate it
tx._handle_feedback_data('buckaroo', buckaroo_post_data)
tx._handle_feedback_data('buckaroo', notification_data)
self.assertEqual(tx.state, 'done')
self.assertEqual(tx.acquirer_reference, notification_data.get('brq_transactions'))
tx._handle_feedback_data('buckaroo', notification_data)
self.assertEqual(tx.state, 'done', 'Buckaroo: validation did not put tx into done state')
self.assertEqual(tx.acquirer_reference, buckaroo_post_data.get('BRQ_TRANSACTIONS'),
'Buckaroo: validation did not update tx payid')
self.assertEqual(tx.acquirer_reference, notification_data.get('brq_transactions'))
# New reference for new tx
self.reference = 'SO004-2'
self.reference = 'Test Transaction 2'
tx = self.create_transaction(flow='redirect')
notification_data = BuckarooController._normalize_data_keys(dict(
self.SYNC_NOTIFICATION_DATA,
brq_invoicenumber=self.reference,
brq_statuscode='2',
brq_signature='b8e54e26b2b5a5e697b8ed5085329ea712fd48b2',
))
self.env['payment.transaction']._handle_feedback_data('buckaroo', notification_data)
self.assertEqual(tx.state, 'error')
buckaroo_post_data['BRQ_INVOICENUMBER'] = self.reference
# now buckaroo post is ok: try to modify the SHASIGN
buckaroo_post_data['Brq_signature'] = '54d928810e343acf5fb0c3ee75fd747ff159ef7a'
with self.assertRaises(ValidationError):
self.env['payment.transaction']._handle_feedback_data('buckaroo', buckaroo_post_data)
def test_accept_notification_with_valid_signature(self):
""" Test the verification of a notification with a valid signature. """
tx = self.create_transaction('redirect')
self._assert_does_not_raise(
Forbidden,
BuckarooController._verify_notification_signature,
self.SYNC_NOTIFICATION_DATA,
self.SYNC_NOTIFICATION_DATA['brq_signature'],
tx,
)
# simulate an error
buckaroo_post_data['BRQ_STATUSCODE'] = '2'
buckaroo_post_data['Brq_signature'] = '3e67da5181b1a895d322987303e42bab2a376eec'
@mute_logger('odoo.addons.payment_buckaroo.controllers.main')
def test_reject_notification_with_missing_signature(self):
""" Test the verification of a notification with a missing signature. """
tx = self.create_transaction('redirect')
self.assertRaises(
Forbidden,
BuckarooController._verify_notification_signature,
self.SYNC_NOTIFICATION_DATA,
None,
tx,
)
# Avoid warning log bc of unknown status code
with mute_logger('odoo.addons.payment_buckaroo.models.payment_transaction'):
self.env['payment.transaction']._handle_feedback_data('buckaroo', buckaroo_post_data)
self.assertEqual(tx.state, 'error',
'Buckaroo: unexpected status code should put tx in error state')
@mute_logger('odoo.addons.payment_buckaroo.controllers.main')
def test_reject_notification_with_invalid_signature(self):
""" Test the verification of a notification with an invalid signature. """
tx = self.create_transaction('redirect')
self.assertRaises(
Forbidden,
BuckarooController._verify_notification_signature,
self.SYNC_NOTIFICATION_DATA,
'dummy',
tx,
)
def test_signature_is_computed_based_on_lower_case_data_keys(self):
""" Test that lower case keys are used to execute the case-insensitive sort. """
addons/payment_mollie/__init__.py
-1
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import controllers
addons/payment_mollie/__manifest__.py
-1
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
{
addons/payment_mollie/controllers/__init__.py
-1
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import main
addons/payment_mollie/controllers/main.py
+9 -10
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
@@ -12,15 +11,15 @@ _logger = logging.getLogger(__name__)
class MollieController(http.Controller):
_return_url = "/payment/mollie/return"
_notify_url = "/payment/mollie/notify"
_return_url = '/payment/mollie/return'
_webhook_url = '/payment/mollie/webhook'
@http.route(
_return_url, type='http', auth='public', methods=['GET', 'POST'], csrf=False,
save_session=False
)
def mollie_return(self, **data):
""" Process the data returned by Mollie after redirection.
def mollie_return_from_checkout(self, **data):
""" Process the notification data sent by Mollie after redirection from checkout.
The route is flagged with `save_session=False` to prevent Odoo from assigning a new session
to the user if they are redirected to this route with a POST request. Indeed, as the session
@@ -37,9 +36,9 @@ class MollieController(http.Controller):
request.env['payment.transaction'].sudo()._handle_feedback_data('mollie', data)
return request.redirect('/payment/status')
@http.route(_notify_url, type='http', auth='public', methods=['POST'], csrf=False)
def mollie_notify(self, **data):
""" Process the data sent by Mollie to the webhook.
@http.route(_webhook_url, type='http', auth='public', methods=['POST'], csrf=False)
def mollie_webhook(self, **data):
""" Process the notification data sent by Mollie to the webhook.
:param dict data: The feedback data (only `id`) and the transaction reference (`ref`)
embedded in the return URL
@@ -50,5 +49,5 @@ class MollieController(http.Controller):
try:
request.env['payment.transaction'].sudo()._handle_feedback_data('mollie', data)
except ValidationError: # Acknowledge the notification to avoid getting spammed
_logger.exception("unable to handle the data; skipping to acknowledge the notification")
return '' # Acknowledge the notification with an HTTP 200 response
_logger.exception("unable to handle the notification data; skipping to acknowledge")
return '' # Acknowledge the notification
addons/payment_mollie/models/__init__.py
-1
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import account_payment_method
addons/payment_mollie/models/account_payment_method.py
-1
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo import api, models
addons/payment_mollie/models/payment_acquirer.py
-1
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
addons/payment_mollie/models/payment_transaction.py
+2 -3
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
@@ -39,7 +38,7 @@ class PaymentTransaction(models.Model):
# The acquirer reference is set now to allow fetching the payment status after redirection
self.acquirer_reference = payment_data.get('id')
return {'api_url': payment_data["_links"]["checkout"]["href"]}
return {'api_url': payment_data['_links']['checkout']['href']}
def _mollie_prepare_payment_request_payload(self):
""" Create the payload for the payment request based on the transaction values.
@@ -50,7 +49,7 @@ class PaymentTransaction(models.Model):
user_lang = self.env.context.get('lang')
base_url = self.acquirer_id.get_base_url()
redirect_url = urls.url_join(base_url, MollieController._return_url)
webhook_url = urls.url_join(base_url, MollieController._notify_url)
webhook_url = urls.url_join(base_url, MollieController._webhook_url)
return {
'description': self.reference,
addons/payment_mollie/tests/common.py
+6 -1
View File
@@ -1,10 +1,15 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.addons.payment.tests.common import PaymentCommon
class MollieCommon(PaymentCommon):
NOTIFICATION_DATA = {
'ref': 'Test Transaction', # Shamefully copy-pasted from payment
'id': 'tr_ABCxyz0123',
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
addons/payment_mollie/tests/test_mollie.py
+24 -4
View File
@@ -1,13 +1,17 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.tests import tagged
from unittest.mock import patch
from .common import MollieCommon
from odoo.tests import tagged
from odoo.tools import mute_logger
from odoo.addons.payment.tests.http_common import PaymentHttpCommon
from odoo.addons.payment_mollie.controllers.main import MollieController
from odoo.addons.payment_mollie.tests.common import MollieCommon
@tagged('post_install', '-at_install')
class MollieTest(MollieCommon):
class MollieTest(MollieCommon, PaymentHttpCommon):
def test_payment_request_payload_values(self):
tx = self.create_transaction(flow='redirect')
@@ -16,3 +20,19 @@ class MollieTest(MollieCommon):
self.assertDictEqual(payload['amount'], {'currency': 'EUR', 'value': '1111.11'})
self.assertEqual(payload['description'], tx.reference)
@mute_logger(
'odoo.addons.payment_mollie.controllers.main',
'odoo.addons.payment_mollie.models.payment_transaction',
)
def test_webhook_notification_confirms_transaction(self):
""" Test the processing of a webhook notification. """
tx = self.create_transaction('redirect')
url = self._build_url(MollieController._webhook_url)
with patch(
'odoo.addons.payment_mollie.models.payment_acquirer.PaymentAcquirer'
'._mollie_make_request',
return_value={'status': 'paid'},
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
addons/payment_ogone/controllers/main.py
+44 -34
View File
@@ -1,13 +1,13 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import hmac
import logging
import pprint
import re
import werkzeug
from werkzeug.exceptions import Forbidden
from odoo import _, http
from odoo.exceptions import ValidationError
from odoo import http
from odoo.http import request
_logger = logging.getLogger(__name__)
@@ -28,55 +28,65 @@ class OgoneController(http.Controller):
@http.route(
[_return_url] + _backward_compatibility_urls, type='http', auth='public',
methods=['GET', 'POST'], csrf=False
) # 'GET' or 'POST' depending on the configuration in Ogone backend
def ogone_return_from_redirect(self, **feedback_data):
""" Process the data returned by Ogone after redirection to the Hosted Payment Page.
) # Redirect are made with GET requests only. Webhook notifications can be set to GET or POST.
def ogone_return_from_checkout(self, **raw_data):
""" Process the notification data sent by Ogone after redirection from checkout.
This route can also accept S2S notifications from Ogone if it is configured as a webhook in
Ogone's backend.
Ogone's backend. The user can choose between GET or POST for the webhook notifications.
:param dict feedback_data: The feedback data
:param dict raw_data: The un-formatted notification data
"""
# Check the source and integrity of the data
data = self._homogenize_data(feedback_data)
self._verify_signature(feedback_data, data)
_logger.info("handling redirection from Ogone with data:\n%s", pprint.pformat(raw_data))
data = self._normalize_data_keys(raw_data)
# Handle the feedback data
_logger.info("handling redirection from Ogone with data:\n%s", pprint.pformat(data))
# Check the integrity of the notification
received_signature = data.get('SHASIGN')
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'ogone', data
)
self._verify_notification_signature(raw_data, received_signature, tx_sudo)
# Handle the notification data
request.env['payment.transaction'].sudo()._handle_feedback_data('ogone', data)
return request.redirect('/payment/status')
def _homogenize_data(self, data):
""" Format keys to follow an homogenized convention inspired by Ogone Directlink API.
@staticmethod
def _normalize_data_keys(data):
""" Set all keys of a dictionary to upper-case.
The keys received from Ogone APIs have inconsistent formatting and must be homogenized to
allow re-using the same methods. We reformat them to follow a unified nomenclature inspired
by DirectLink's order direct endpoint.
by Ogone Directlink API.
Formatting steps:
1) Uppercase key strings: 'Something' -> 'SOMETHING', 'something' -> 'SOMETHING'
2) Remove the prefix: 'CARD.SOMETHING' -> 'SOMETHING', 'ALIAS.SOMETHING' -> 'SOMETHING'
:param dict data: The data whose keys to normalize
:return: The normalized data
:rtype: dict
"""
return {re.sub(r'.*\.', '', k.upper()): v for k, v in data.items()}
def _verify_signature(self, sign_data, data):
""" Check that the signature computed from the feedback matches the received one.
@staticmethod
def _verify_notification_signature(notification_data, received_signature, tx_sudo):
""" Check that the received signature matches the expected one.
:param dict sign_data: The original feedback data used to compute the signature
:param dict sign_data: The formatted feedback data used to find the tx and received sig
:param dict notification_data: The notification data
:param str received_signature: The signature received with the notification data
:param recordset tx_sudo: The sudoed transaction referenced by the notification data, as a
`payment.transaction` record
:return: None
:raise: ValidationError if the signatures don't match
:raise: :class:`werkzeug.exceptions.Forbidden` if the signatures don't match
"""
acquirer_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'ogone', data
).acquirer_id # Find the acquirer based on the transaction
received_signature = data.get('SHASIGN')
expected_signature = acquirer_sudo._ogone_generate_signature(sign_data)
if received_signature != expected_signature.upper():
raise ValidationError(
"Ogone: " + _(
"Received data with invalid signature. expected: %(exp)s ; received: %(rec)s ; "
"data:\n%(data)s",
exp=expected_signature, rec=received_signature, data=pprint.pformat(sign_data)
)
)
# Check for the received signature
if not received_signature:
_logger.warning("received notification with missing signature")
raise Forbidden()
# Compare the received signature with the expected signature computed from the data
expected_signature = tx_sudo.acquirer_id._ogone_generate_signature(notification_data)
if not hmac.compare_digest(received_signature, expected_signature.upper()):
_logger.warning("received notification with invalid signature")
raise Forbidden()
addons/payment_ogone/models/payment_acquirer.py
+1 -2
View File
@@ -88,8 +88,7 @@ class PaymentAcquirer(models.Model):
formatted_items = [(k.upper(), v) for k, v in values.items()]
sorted_items = sorted(formatted_items)
signing_string = ''.join(f'{k}={v}{key}' for k, v in sorted_items if _filter_key(k) and v)
shasign = sha1(signing_string.encode()).hexdigest()
return shasign
return sha1(signing_string.encode()).hexdigest()
def _ogone_make_request(self, payload=None, method='POST'):
""" Make a request to one of Ogone APIs.
addons/payment_ogone/tests/common.py
+16
View File
@@ -5,6 +5,22 @@ from odoo.addons.payment.tests.common import PaymentCommon
class OgoneCommon(PaymentCommon):
NOTIFICATION_DATA = {
'AAVADDRESS': 'NO',
'amount': '1111.11',
'CARDNO': 'XXXXXXXXXXXX1111',
'CN': 'Dummy Customer Name',
'currency': 'USD',
'IP': '101.00.111.22',
'NCERROR': '0',
'orderID': 'Test Transaction', # Shamefully copy-pasted from payment
'PAYID': '01234567899',
'PM': 'CreditCard',
'SHASIGN': '2CE444D2260D914EA7E56450B7B28F189238553B',
'STATUS': '9', # 'Payment requested' (done)
'TRXDATE': '01/31/22',
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
addons/payment_ogone/tests/test_ogone.py
+69 -4
View File
@@ -1,19 +1,22 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from unittest.mock import patch
from freezegun import freeze_time
from werkzeug.exceptions import Forbidden
from odoo.fields import Command
from odoo.tests import tagged
from odoo.tools import mute_logger
from odoo.addons.payment import utils as payment_utils
from .common import OgoneCommon
from ..controllers.main import OgoneController
from odoo.addons.payment.tests.http_common import PaymentHttpCommon
from odoo.addons.payment_ogone.controllers.main import OgoneController
from odoo.addons.payment_ogone.tests.common import OgoneCommon
@tagged('post_install', '-at_install')
class OgoneTest(OgoneCommon):
class OgoneTest(OgoneCommon, PaymentHttpCommon):
def test_incompatibility_with_validation_operation(self):
acquirers = self.env['payment.acquirer']._get_compatible_acquirers(
@@ -96,3 +99,65 @@ class OgoneTest(OgoneCommon):
value,
f"received value {inputs[form_key]} for input {form_key} (expected {value})"
)
@mute_logger('odoo.addons.payment_ogone.controllers.main')
def test_webhook_notification_confirms_transaction(self):
""" Test the processing of a webhook notification. """
tx = self.create_transaction('redirect')
url = self._build_url(OgoneController._return_url)
with patch(
'odoo.addons.payment_ogone.controllers.main.OgoneController'
'._verify_notification_signature'
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
@mute_logger('odoo.addons.payment_ogone.controllers.main')
def test_webhook_notification_triggers_signature_check(self):
""" Test that receiving a webhook notification triggers a signature check. """
self.create_transaction('redirect')
url = self._build_url(OgoneController._return_url)
with patch(
'odoo.addons.payment_ogone.controllers.main.OgoneController'
'._verify_notification_signature'
) as signature_check_mock, patch(
'odoo.addons.payment.models.payment_transaction.PaymentTransaction'
'._handle_feedback_data'
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(signature_check_mock.call_count, 1)
def test_accept_notification_with_valid_signature(self):
""" Test the verification of a notification with a valid signature. """
tx = self.create_transaction('redirect')
self._assert_does_not_raise(
Forbidden,
OgoneController._verify_notification_signature,
self.NOTIFICATION_DATA,
self.NOTIFICATION_DATA['SHASIGN'],
tx,
)
@mute_logger('odoo.addons.payment_ogone.controllers.main')
def test_reject_notification_with_missing_signature(self):
""" Test the verification of a notification with a missing signature. """
tx = self.create_transaction('redirect')
self.assertRaises(
Forbidden,
OgoneController._verify_notification_signature,
self.NOTIFICATION_DATA,
None,
tx,
)
@mute_logger('odoo.addons.payment_ogone.controllers.main')
def test_reject_notification_with_invalid_signature(self):
""" Test the verification of a notification with an invalid signature. """
tx = self.create_transaction('redirect')
self.assertRaises(
Forbidden,
OgoneController._verify_notification_signature,
self.NOTIFICATION_DATA,
'dummy',
tx,
)
addons/payment_paypal/controllers/main.py
+78 -52
View File
@@ -4,8 +4,9 @@ import logging
import pprint
import requests
from werkzeug.exceptions import Forbidden
from odoo import _, http
from odoo import http
from odoo.exceptions import ValidationError
from odoo.http import request
@@ -13,18 +14,22 @@ _logger = logging.getLogger(__name__)
class PaypalController(http.Controller):
_return_url = '/payment/paypal/dpn/'
_notify_url = '/payment/paypal/ipn/'
_return_url = '/payment/paypal/return/'
_webhook_url = '/payment/paypal/webhook/'
@http.route(
_return_url, type='http', auth='public', methods=['GET', 'POST'], csrf=False,
save_session=False
)
def paypal_dpn(self, **data):
""" Route used by the PDT notification.
def paypal_return_from_checkout(self, **data):
""" Process the notification data (PDT) sent by PayPal after redirection from checkout.
The "PDT notification" is actually POST data sent along the user redirection.
The route also allows the GET method in case the user clicks on "go back to merchant site".
The "Payment Data Transfer" notification is actually the notification data sent along the
redirect. See https://developer.paypal.com/api/nvp-soap/payment-data-transfer/.
The route accept both GET and POST requests because PayPal seems to switch between the two
depending on whether PDT is enabled, the customer pays anonymously (without logging in on
PayPal), the customer clicks on "go back to merchant site" and cancels the payment, etc.
The route is flagged with `save_session=False` to prevent Odoo from assigning a new session
to the user if they are redirected to this route with a POST request. Indeed, as the session
@@ -34,38 +39,57 @@ class PaypalController(http.Controller):
will satisfy any specification of the `SameSite` attribute, the session of the user will be
retrieved and with it the transaction which will be immediately post-processed.
"""
_logger.info("handling redirection from Ogone with data:\n%s", pprint.pformat(data))
try:
self._validate_data_authenticity(**data)
except ValidationError:
pass # The transaction has been moved to state 'error'. Redirect to /payment/status.
else:
if data:
request.env['payment.transaction'].sudo()._handle_feedback_data('paypal', data)
else:
pass # The customer has cancelled the payment, don't do anything
_logger.info("handling redirection from PayPal with data:\n%s", pprint.pformat(data))
# Check the origin of the notification
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'paypal', data
)
self._verify_notification_origin(data, tx_sudo)
# Handle the notification data
if data:
request.env['payment.transaction'].sudo()._handle_feedback_data('paypal', data)
else: # The customer has cancelled the payment
pass # Redirect the customer to the status page
return request.redirect('/payment/status')
@http.route(_notify_url, type='http', auth='public', methods=['GET', 'POST'], csrf=False)
def paypal_ipn(self, **data):
""" Route used by the IPN. """
_logger.info("notification received from Ogone with data:\n%s", pprint.pformat(data))
@http.route(_webhook_url, type='http', auth='public', methods=['GET', 'POST'], csrf=False)
def paypal_webhook(self, **data):
""" Process the notification data (IPN) sent by PayPal to the webhook.
The "Instant Payment Notification" is a classical webhook notification.
See https://developer.paypal.com/api/nvp-soap/ipn/.
:param dict data: The notification data
:return: An empty string to acknowledge the notification
:rtype: str
"""
_logger.info("notification received from PayPal with data:\n%s", pprint.pformat(data))
try:
self._validate_data_authenticity(**data)
# Check the origin and integrity of the notification
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'paypal', data
)
self._verify_notification_origin(data, tx_sudo)
# Handle the notification data
request.env['payment.transaction'].sudo()._handle_feedback_data('paypal', data)
except ValidationError: # Acknowledge the notification to avoid getting spammed
_logger.exception("unable to handle the data; skipping to acknowledge the notification")
_logger.exception("unable to handle the notification data; skipping to acknowledge")
return ''
def _validate_data_authenticity(self, **data):
""" Validate the authenticity of data received through DPN or IPN
@staticmethod
def _verify_notification_origin(notification_data, tx_sudo):
""" Check that the notification was sent by Alipay.
The verification is done in three steps:
- 1: POST the complete, unaltered, message back to Paypal (preceded by
`cmd=_notify-validate`), in the same encoding.
- 1: POST the complete message back to Paypal with the additional param
`cmd=_notify-validate`, in the same encoding.
- 2: PayPal sends back either 'VERIFIED' or 'INVALID'.
- 3: Return an empty HTTP 200 response (done at the end of the route method).
See https://developer.paypal.com/docs/api-basics/notifications/ipn/IPNIntro
- 3: Return an empty HTTP 200 response if the notification origin is verified by PayPal,
raise an HTTP 403 otherwise.
See https://developer.paypal.com/api/nvp-soap/ipn/IPNImplementation/
As per https://developer.paypal.com/docs/api-basics/notifications/payment-data-transfer/,
PDT notifications should be verified in a similar but different manner:
@@ -82,32 +106,34 @@ class PaypalController(http.Controller):
Since PDT notifications have in practice always been successfully authenticated by using the
IPN protocol, this method does explicitly that for both PDT and IPN.
:param dict data: The data whose authenticity to check
:param dict notification_data: The notification data
:param recordset tx_sudo: The sudoed transaction referenced in the notification data, as a
`payment.transaction` record
:return: None
:raise: ValidationError if the authenticity could not be verified
:raise: :class:`werkzeug.exceptions.Forbidden` if the notification origin can't be verified
"""
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'paypal', data
)
acquirer_sudo = tx_sudo.acquirer_id
# Request PayPal for an authenticity check
data['cmd'] = '_notify-validate'
response = requests.post(acquirer_sudo._paypal_get_api_url(), data, timeout=60)
response.raise_for_status()
# Inspect the response code and raise if not 'VERIFIED'.
response_code = response.text
if response_code == 'VERIFIED':
_logger.info("authenticity of notification data verified")
url = tx_sudo.acquirer_id._paypal_get_api_url()
payload = dict(notification_data, cmd='_notify-validate')
try:
response = requests.post(url, payload, timeout=60)
response.raise_for_status()
except (requests.exceptions.ConnectionError, requests.exceptions.HTTPError) as error:
_logger.exception(
"could not verify notification origin at %(url)s with data: %(data)s:\n%(error)s",
{
'url': url,
'data': pprint.pformat(notification_data),
'error': pprint.pformat(error.response.text),
},
)
raise Forbidden()
else:
if response_code == 'INVALID':
error_message = "PayPal: " + _("Notification data were not acknowledged.")
else:
error_message = "PayPal: " + _(
"Received unrecognized authentication check response code: received %s, "
"expected VERIFIED or INVALID.",
response_code
response_content = response.text
if response_content != 'VERIFIED':
_logger.error(
"PayPal did not confirm the origin of the notification with data:\n%s",
pprint.pformat(notification_data),
)
tx_sudo._set_error(error_message)
raise ValidationError(error_message)
raise Forbidden()
addons/payment_paypal/models/payment_transaction.py
+2 -3
View File
@@ -36,8 +36,7 @@ class PaymentTransaction(models.Model):
base_url = self.acquirer_id.get_base_url()
partner_first_name, partner_last_name = payment_utils.split_partner_name(self.partner_name)
notify_url = self.acquirer_id.paypal_use_ipn \
and urls.url_join(base_url, PaypalController._notify_url)
webhook_url = urls.url_join(base_url, PaypalController._webhook_url)
return {
'address1': self.partner_address,
'amount': self.amount,
@@ -52,7 +51,7 @@ class PaymentTransaction(models.Model):
'item_number': self.reference,
'last_name': partner_last_name,
'lc': self.partner_lang,
'notify_url': notify_url,
'notify_url': webhook_url if self.acquirer_id.paypal_use_ipn else None,
'return_url': urls.url_join(base_url, PaypalController._return_url),
'state': self.partner_state_id.name,
'zip_code': self.partner_zip,
addons/payment_paypal/tests/common.py
+36
View File
@@ -1,9 +1,45 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.addons.payment.tests.common import PaymentCommon
class PaypalCommon(PaymentCommon):
NOTIFICATION_DATA = {
'PayerID': '59XDVNACRAZZK',
'address_city': 'Scranton',
'address_country_code': 'US',
'address_name': 'Mitchell Admin',
'address_state': 'Pennsylvania',
'address_street': '215 Vine St',
'address_zip': '18503',
'first_name': 'Norbert',
'handling_amount': '0.00',
'item_name': 'YourCompany: Test Transaction',
'item_number': 'Test Transaction', # Shamefully copy-pasted from payment
'last_name': 'Buyer',
'mc_currency': 'USD',
'mc_fee': '2.00',
'mc_gross': '50.00',
'notify_version': 'UNVERSIONED',
'payer_email': 'test-buyer@mail.odoo.com',
'payer_id': '59XDVNACRAZZK',
'payer_status': 'VERIFIED',
'payment_date': '2022-01-19T08:38:06Z',
'payment_fee': '2.00',
'payment_gross': '50.00',
'payment_status': 'Completed',
'payment_type': 'instant',
'protection_eligibility': 'ELIGIBLE',
'quantity': '1',
'receiver_id': 'BEQE89VH6257B',
'residence_country': 'US',
'shipping': '0.00',
'txn_id': '1H89255869624041K',
'txn_type': 'web_accept',
'verify_sign': 'dummy',
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
addons/payment_paypal/tests/test_paypal.py
+59 -70
View File
@@ -1,15 +1,18 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.exceptions import ValidationError
from odoo.tools import mute_logger
from odoo.tests import tagged
from unittest.mock import patch
from .common import PaypalCommon
from ..controllers.main import PaypalController
from odoo.exceptions import ValidationError
from odoo.tests import tagged
from odoo.tools import mute_logger
from odoo.addons.payment.tests.http_common import PaymentHttpCommon
from odoo.addons.payment_paypal.controllers.main import PaypalController
from odoo.addons.payment_paypal.tests.common import PaypalCommon
@tagged('post_install', '-at_install')
class PaypalForm(PaypalCommon):
class PaypalTest(PaypalCommon, PaymentHttpCommon):
def _get_expected_values(self):
return_url = self._build_url(PaypalController._return_url)
@@ -28,7 +31,7 @@ class PaypalForm(PaypalCommon):
'item_number': self.reference,
'last_name': 'Buyer',
'lc': 'en_US',
'notify_url': self._build_url(PaypalController._notify_url),
'notify_url': self._build_url(PaypalController._webhook_url),
'return': return_url,
'rm': '2',
'zip': '1000',
@@ -78,74 +81,33 @@ class PaypalForm(PaypalCommon):
"Paypal: invalid inputs specified in the redirect form.")
def test_feedback_processing(self):
# typical data posted by paypal after client has successfully paid
paypal_post_data = {
'protection_eligibility': u'Ineligible',
'last_name': u'Poilu',
'txn_id': u'08D73520KX778924N',
'receiver_email': 'dummy',
'payment_status': u'Pending',
'payment_gross': u'',
'tax': u'0.00',
'residence_country': u'FR',
'address_state': u'Alsace',
'payer_status': u'verified',
'txn_type': u'web_accept',
'address_street': u'Av. de la Pelouse, 87648672 Mayet',
'handling_amount': u'0.00',
'payment_date': u'03:21:19 Nov 18, 2013 PST',
'first_name': u'Norbert',
'item_name': self.reference,
'address_country': u'France',
'charset': u'windows-1252',
'notify_version': u'3.7',
'address_name': u'Norbert Poilu',
'pending_reason': u'multi_currency',
'item_number': self.reference,
'receiver_id': u'dummy',
'transaction_subject': u'',
'business': u'dummy',
'test_ipn': u'1',
'payer_id': u'VTDKRZQSAHYPS',
'verify_sign': u'An5ns1Kso7MWUdW4ErQKJJJ4qi4-AVoiUf-3478q3vrSmqh08IouiYpM',
'address_zip': u'75002',
'address_country_code': u'FR',
'address_city': u'Paris',
'address_status': u'unconfirmed',
'mc_currency': u'EUR',
'shipping': u'0.00',
'payer_email': u'tde+buyer@odoo.com',
'payment_type': u'instant',
'mc_gross': str(self.amount),
'ipn_track_id': u'866df2ccd444b',
'quantity': u'1'
}
# should raise error about unknown tx
# Unknown transaction
with self.assertRaises(ValidationError):
self.env['payment.transaction']._handle_feedback_data('paypal', paypal_post_data)
self.env['payment.transaction']._handle_feedback_data('paypal', self.NOTIFICATION_DATA)
tx = self.create_transaction(flow='redirect')
# Confirmed transaction
tx = self.create_transaction('redirect')
self.env['payment.transaction']._handle_feedback_data('paypal', self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
self.assertEqual(tx.acquirer_reference, self.NOTIFICATION_DATA['txn_id'])
# Validate the transaction (pending feedback)
self.env['payment.transaction']._handle_feedback_data('paypal', paypal_post_data)
self.assertEqual(tx.state, 'pending', 'paypal: wrong state after receiving a valid pending notification')
self.assertEqual(tx.state_message, 'multi_currency', 'paypal: wrong state message after receiving a valid pending notification')
self.assertEqual(tx.acquirer_reference, '08D73520KX778924N', 'paypal: wrong txn_id after receiving a valid pending notification')
# Reset the transaction
tx.write({'state': 'draft', 'acquirer_reference': False})
# Validate the transaction ('completed' feedback)
paypal_post_data['payment_status'] = 'Completed'
self.env['payment.transaction']._handle_feedback_data('paypal', paypal_post_data)
self.assertEqual(tx.state, 'done', 'paypal: wrong state after receiving a valid pending notification')
self.assertEqual(tx.acquirer_reference, '08D73520KX778924N', 'paypal: wrong txn_id after receiving a valid pending notification')
# Pending transaction
self.reference = 'Test Transaction 2'
tx = self.create_transaction('redirect')
payload = dict(
self.NOTIFICATION_DATA,
item_number=self.reference,
payment_status='Pending',
pending_reason='multi_currency',
)
self.env['payment.transaction']._handle_feedback_data('paypal', payload)
self.assertEqual(tx.state, 'pending')
self.assertEqual(tx.state_message, payload['pending_reason'])
def test_fees_computation(self):
#If the merchant needs to keep 100€, the transaction will be equal to 103.30€.
#In this way, Paypal will take 103.30 * 2.9% + 0.30 = 3.30€
#And the merchant will take 103.30 - 3.30 = 100€
# If the merchant needs to keep 100€, the transaction will be equal to 103.30€.
# In this way, Paypal will take 103.30 * 2.9% + 0.30 = 3.30€
# And the merchant will take 103.30 - 3.30 = 100€
self.paypal.write({
'fees_active': True,
'fees_int_fixed': 0.30,
@@ -153,3 +115,30 @@ class PaypalForm(PaypalCommon):
})
total_fee = self.paypal._compute_fees(100, False, False)
self.assertEqual(round(total_fee, 2), 3.3, 'Wrong computation of the Paypal fees')
@mute_logger('odoo.addons.payment_paypal.controllers.main')
def test_webhook_notification_confirms_transaction(self):
""" Test the processing of a webhook notification. """
tx = self.create_transaction('redirect')
url = self._build_url(PaypalController._webhook_url)
with patch(
'odoo.addons.payment_paypal.controllers.main.PaypalController'
'._verify_notification_origin'
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
@mute_logger('odoo.addons.payment_paypal.controllers.main')
def test_webhook_notification_triggers_origin_check(self):
""" Test that receiving a webhook notification triggers an origin check. """
self.create_transaction('redirect')
url = self._build_url(PaypalController._webhook_url)
with patch(
'odoo.addons.payment_paypal.controllers.main.PaypalController'
'._verify_notification_origin'
) as origin_check_mock, patch(
'odoo.addons.payment.models.payment_transaction.PaymentTransaction'
'._handle_feedback_data'
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(origin_check_mock.call_count, 1)
addons/payment_payulatam/models/payment_transaction.py
+2 -1
View File
@@ -1,5 +1,6 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import hmac
import logging
from werkzeug import urls
@@ -117,7 +118,7 @@ class PaymentTransaction(models.Model):
# Verify signature
sign_check = tx.acquirer_id._payulatam_generate_sign(data, incoming=True)
if sign_check != sign:
if not hmac.compare_digest(sign_check, sign):
raise ValidationError(
"PayU Latam: " + _(
"Invalid sign: received %(sign)s, computed %(check)s.",
addons/payment_payumoney/controllers/main.py
+40 -3
View File
@@ -1,8 +1,11 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import hmac
import logging
import pprint
from werkzeug.exceptions import Forbidden
from odoo import http
from odoo.http import request
@@ -16,8 +19,10 @@ class PayUMoneyController(http.Controller):
_return_url, type='http', auth='public', methods=['GET', 'POST'], csrf=False,
save_session=False
)
def payumoney_return(self, **data):
""" Process the data returned by PayUmoney after redirection.
def payumoney_return_from_checkout(self, **data):
""" Process the notification data sent by PayUmoney after redirection from checkout.
See https://developer.payumoney.com/redirect/.
The route is flagged with `save_session=False` to prevent Odoo from assigning a new session
to the user if they are redirected to this route with a POST request. Indeed, as the session
@@ -27,8 +32,40 @@ class PayUMoneyController(http.Controller):
will satisfy any specification of the `SameSite` attribute, the session of the user will be
retrieved and with it the transaction which will be immediately post-processed.
:param dict data: The feedback data to process
:param dict data: The notification data
"""
_logger.info("handling redirection from PayU money with data:\n%s", pprint.pformat(data))
# Check the integrity of the notification
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'payumoney', data
)
self._verify_notification_signature(data, tx_sudo)
# Handle the notification data
request.env['payment.transaction'].sudo()._handle_feedback_data('payumoney', data)
return request.redirect('/payment/status')
@staticmethod
def _verify_notification_signature(notification_data, tx_sudo):
""" Check that the received signature matches the expected one.
:param dict notification_data: The notification data
:param recordset tx_sudo: The sudoed transaction referenced by the notification data, as a
`payment.transaction` record
:return: None
:raise: :class:`werkzeug.exceptions.Forbidden` if the signatures don't match
"""
# Retrieve the received signature from the data
received_signature = notification_data.get('hash')
if not received_signature:
_logger.warning("received notification with missing signature")
raise Forbidden()
# Compare the received signature with the expected signature computed from the data
expected_signature = tx_sudo.acquirer_id._payumoney_generate_sign(
notification_data, incoming=True
)
if not hmac.compare_digest(received_signature, expected_signature):
_logger.warning("received notification with invalid signature")
raise Forbidden()
addons/payment_payumoney/models/payment_transaction.py
+2 -17
View File
@@ -55,20 +55,15 @@ class PaymentTransaction(models.Model):
:rtype: recordset of `payment.transaction`
:raise: ValidationError if inconsistent data were received
:raise: ValidationError if the data match no transaction
:raise: ValidationError if the signature can not be verified
"""
tx = super()._get_tx_from_feedback_data(provider, data)
if provider != 'payumoney':
return tx
reference = data.get('txnid')
shasign = data.get('hash')
if not reference or not shasign:
if not reference:
raise ValidationError(
"PayUmoney: " + _(
"Received data with missing reference (%(ref)s) or shasign (%(sign)s)",
ref=reference, sign=shasign,
)
"PayUmoney: " + _("Received data with missing reference (%s)", reference)
)
tx = self.search([('reference', '=', reference), ('provider', '=', 'payumoney')])
@@ -77,16 +72,6 @@ class PaymentTransaction(models.Model):
"PayUmoney: " + _("No transaction found matching reference %s.", reference)
)
# Verify shasign
shasign_check = tx.acquirer_id._payumoney_generate_sign(data, incoming=True)
if shasign_check != shasign:
raise ValidationError(
"PayUmoney: " + _(
"Invalid shasign: received %(sign)s, computed %(computed)s.",
sign=shasign, computed=shasign_check
)
)
return tx
def _process_feedback_data(self, data):
addons/payment_payumoney/tests/common.py
+48
View File
@@ -1,9 +1,57 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.addons.payment.tests.common import PaymentCommon
class PayumoneyCommon(PaymentCommon):
NOTIFICATION_DATA = {
'PG_TYPE': 'HDFCPG',
'addedon': '2022-01-19 21:36:05',
'address1': '',
'address2': '',
'amount': '100.00',
'amount_split': '{"PAYU":"100.00"}',
'bank_ref_num': '429387287430473',
'bankcode': 'VISA',
'cardhash': 'This field is no longer supported in postback params.',
'cardnum': '401200XXXXXX1112',
'city': '',
'country': '',
'discount': '0.00',
'email': 'admin@yourcompany.example.com',
'encryptedPaymentId': 'A45B5A1E189A9FC89C0E150F2E6EF074',
'error': 'E000',
'error_Message': 'No Error',
'field1': '021268586345',
'field2': '315728',
'field3': '429387287430473',
'field4': 'ejRWR1J6RFRQWnd1c1BXb2FTbVk=',
'field5': '05',
'field6': '',
'field7': 'AUTHPOSITIVE',
'field8': '',
'field9': '',
'firstname': 'Mitchell',
'giftCardIssued': 'true',
'hash': '6618df5167d46785efeb0ef392d9a150de0c6e56699eb67898ebc690202fa5fe53cee1290b2ca2dfdc307e5bff2518e30dfa680923b538d7ab7f1624a4a9baa5',
'isConsentPayment': '0',
'key': 'GuWsdB4T',
'lastname': '',
'mihpayid': '9084338127',
'mode': 'CC',
'name_on_card': 'ABC DEF',
'net_amount_debit': '100',
'payuMoneyId': '251115271',
'phone': '5555555555',
'productinfo': 'anv-test-20220118140529',
'state': '',
'status': 'success',
'txnid': 'Test Transaction', # Shamefully copy-pasted from payment
'unmappedstatus': 'captured',
'zipcode': '',
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
addons/payment_payumoney/tests/test_payumoney.py
+34 -3
View File
@@ -1,15 +1,18 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from werkzeug.exceptions import Forbidden
from odoo.tests import tagged
from odoo.tools import mute_logger
from .common import PayumoneyCommon
from ..controllers.main import PayUMoneyController
from odoo.addons.payment import utils as payment_utils
from odoo.addons.payment.tests.http_common import PaymentHttpCommon
from odoo.addons.payment_payumoney.controllers.main import PayUMoneyController
from odoo.addons.payment_payumoney.tests.common import PayumoneyCommon
@tagged('post_install', '-at_install')
class PayumoneyTest(PayumoneyCommon):
class PayUMoneyTest(PayumoneyCommon, PaymentHttpCommon):
def test_compatible_acquirers(self):
acquirers = self.env['payment.acquirer']._get_compatible_acquirers(
@@ -54,3 +57,31 @@ class PayumoneyTest(PayumoneyCommon):
'https://sandboxsecure.payu.in/_payment')
self.assertDictEqual(form_info['inputs'], expected_values,
"PayUMoney: invalid inputs specified in the redirect form.")
def test_accept_notification_with_valid_signature(self):
""" Test the verification of a notification with a valid signature. """
tx = self.create_transaction('redirect')
self._assert_does_not_raise(
Forbidden,
PayUMoneyController._verify_notification_signature,
self.NOTIFICATION_DATA,
tx,
)
@mute_logger('odoo.addons.payment_payumoney.controllers.main')
def test_reject_notification_with_missing_signature(self):
""" Test the verification of a notification with a missing signature. """
tx = self.create_transaction('redirect')
payload = dict(self.NOTIFICATION_DATA, hash=None)
self.assertRaises(
Forbidden, PayUMoneyController._verify_notification_signature, payload, tx
)
@mute_logger('odoo.addons.payment_payumoney.controllers.main')
def test_reject_notification_with_invalid_signature(self):
""" Test the verification of a notification with an invalid signature. """
tx = self.create_transaction('redirect')
payload = dict(self.NOTIFICATION_DATA, hash='dummy')
self.assertRaises(
Forbidden, PayUMoneyController._verify_notification_signature, payload, tx
)
addons/payment_sips/controllers/main.py
+59 -43
View File
@@ -1,9 +1,12 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
# Original Copyright 2015 Eezee-It, modified and maintained by Odoo.
import hmac
import logging
import pprint
from werkzeug.exceptions import Forbidden
from odoo import http
from odoo.exceptions import ValidationError
from odoo.http import request
@@ -12,14 +15,14 @@ _logger = logging.getLogger(__name__)
class SipsController(http.Controller):
_return_url = '/payment/sips/dpn/'
_notify_url = '/payment/sips/ipn/'
_return_url = '/payment/sips/return/'
_webhook_url = '/payment/sips/webhook/'
@http.route(
_return_url, type='http', auth='public', methods=['POST'], csrf=False, save_session=False
)
def sips_dpn(self, **post):
""" Process the data returned by SIPS after redirection.
def sips_return_from_checkout(self, **data):
""" Process the notification data sent by SIPS after redirection from checkout.
The route is flagged with `save_session=False` to prevent Odoo from assigning a new session
to the user if they are redirected to this route with a POST request. Indeed, as the session
@@ -29,47 +32,60 @@ class SipsController(http.Controller):
will satisfy any specification of the `SameSite` attribute, the session of the user will be
retrieved and with it the transaction which will be immediately post-processed.
:param dict post: The feedback data to process
:param dict data: The notification data
"""
_logger.info("handling redirection from SIPS with data:\n%s", pprint.pformat(post))
try:
if self._sips_validate_data(post):
request.env['payment.transaction'].sudo()._handle_feedback_data('sips', post)
except ValidationError:
pass
_logger.info("handling redirection from SIPS with data:\n%s", pprint.pformat(data))
# Check the integrity of the notification
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'sips', data
)
self._verify_notification_signature(data, tx_sudo)
# Handle the notification data
request.env['payment.transaction'].sudo()._handle_feedback_data('sips', data)
return request.redirect('/payment/status')
@http.route(_notify_url, type='http', auth='public', methods=['POST'], csrf=False)
def sips_ipn(self, **post):
""" Sips IPN. """
_logger.info("notification received from SIPS with data:\n%s", pprint.pformat(post))
if not post:
# SIPS sometimes sends empty notifications, the reason why is unclear but they tend to
# pollute logs and do not provide any meaningful information; log as a warning instead
# of a traceback.
_logger.warning("unable to handle the data; skipping to acknowledge the notification")
else:
try:
if self._sips_validate_data(post):
request.env['payment.transaction'].sudo()._handle_feedback_data('sips', post)
except ValidationError:
pass # Acknowledge the notification to avoid getting spammed
@http.route(_webhook_url, type='http', auth='public', methods=['POST'], csrf=False)
def sips_webhook(self, **data):
""" Process the notification data sent by SIPS to the webhook.
:param dict data: The notification data
:return: An empty string to acknowledge the notification
:rtype: str
"""
_logger.info("notification received from SIPS with data:\n%s", pprint.pformat(data))
try:
# Check the integrity of the notification
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'sips', data
)
self._verify_notification_signature(data, tx_sudo)
# Handle the notification data
request.env['payment.transaction'].sudo()._handle_feedback_data('sips', data)
except ValidationError:
_logger.exception("unable to handle the notification data; skipping to acknowledge")
return ''
def _sips_validate_data(self, post):
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data('sips', post)
acquirer_sudo = tx_sudo.acquirer_id
security = acquirer_sudo._sips_generate_shasign(post['Data'])
if security == post['Seal']:
_logger.debug(
"authenticity of notification data verified for transaction with reference %s",
tx_sudo.reference
)
return True
else:
_logger.warning(
"unable to verify the authenticity of notification data for transaction with "
"reference %s",
tx_sudo.reference,
)
return False
@staticmethod
def _verify_notification_signature(notification_data, tx_sudo):
""" Check that the received signature matches the expected one.
:param dict notification_data: The notification data
:param recordset tx_sudo: The sudoed transaction referenced by the notification data, as a
`payment.transaction` record
:return: None
:raise: :class:`werkzeug.exceptions.Forbidden` if the signatures don't match
"""
# Retrieve the received signature from the data
received_signature = notification_data.get('Seal')
if not received_signature:
_logger.warning("received notification with missing signature")
raise Forbidden()
# Compare the received signature with the expected signature computed from the data
expected_signature = tx_sudo.acquirer_id._sips_generate_shasign(notification_data['Data'])
if not hmac.compare_digest(received_signature, expected_signature):
_logger.warning("received notification with invalid signature")
raise Forbidden()
addons/payment_sips/models/payment_transaction.py
+1 -19
View File
@@ -66,7 +66,7 @@ class PaymentTransaction(models.Model):
'currencyCode': SUPPORTED_CURRENCIES[self.currency_id.name], # The ISO 4217 code
'merchantId': self.acquirer_id.sips_merchant_id,
'normalReturnUrl': urls.url_join(base_url, SipsController._return_url),
'automaticResponseUrl': urls.url_join(base_url, SipsController._notify_url),
'automaticResponseUrl': urls.url_join(base_url, SipsController._webhook_url),
'transactionReference': self.reference,
'statementReference': self.reference,
'keyVersion': self.acquirer_id.sips_key_version,
@@ -91,8 +91,6 @@ class PaymentTransaction(models.Model):
:return: The transaction if found
:rtype: recordset of `payment.transaction`
:raise: ValidationError if the data match no transaction
:raise: ValidationError if the currency is not supported
:raise: ValidationError if the amount mismatch
"""
tx = super()._get_tx_from_feedback_data(provider, data)
if provider != 'sips':
@@ -111,22 +109,6 @@ class PaymentTransaction(models.Model):
"Sips: " + _("No transaction found matching reference %s.", reference)
)
sips_currency = SUPPORTED_CURRENCIES.get(tx.currency_id.name)
if not sips_currency:
raise ValidationError(
"Sips: " + _("This currency is not supported: %s.", tx.currency_id.name)
)
amount_converted = payment_utils.to_major_currency_units(
float(data.get('amount', '0.0')), tx.currency_id
)
if tx.currency_id.compare_amounts(amount_converted, tx.amount) != 0:
raise ValidationError(
"Sips: " + _(
"Incorrect amount: received %(received).2f, expected %(expected).2f",
received=amount_converted, expected=tx.amount
)
)
return tx
def _process_feedback_data(self, data):
addons/payment_sips/tests/common.py
+23
View File
@@ -1,9 +1,32 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.addons.payment.tests.common import PaymentCommon
class SipsCommon(PaymentCommon):
NOTIFICATION_DATA = {
'Data': 'captureDay=0|captureMode=AUTHOR_CAPTURE|currencyCode=840'
'|merchantId=002001000000001|orderChannel=INTERNET|responseCode=00'
'|transactionDateTime=2022-01-19T18:01:06+01:00'
'|transactionReference=Test Transaction' # Shamefully copy-pasted from payment
'|keyVersion=1|acquirerResponseCode=00|amount=10000|authorisationId=12345'
'|guaranteeIndicator=Y|cardCSCResultCode=4D|panExpiryDate=202201'
'|paymentMeanBrand=VISA|paymentMeanType=CARD|customerIpAddress=111.11.111.11'
'|maskedPan=4100##########00|returnContext={"reference": "Test Transaction"}'
'|holderAuthentRelegation=N|holderAuthentStatus=3D_SUCCESS'
'|tokenPan=dp528b9xwknujmkw|transactionOrigin=INTERNET|paymentPattern=ONE_SHOT'
'|customerMobilePhone=null|mandateAuthentMethod=null|mandateUsage=null'
'|transactionActors=null|mandateId=null|captureLimitDate=20220119|dccStatus=null'
'|dccResponseCode=null|dccAmount=null|dccCurrencyCode=null|dccExchangeRate=null'
'|dccExchangeRateValidity=null|dccProvider=null|statementReference=tx20220119170050'
'|panEntryMode=MANUAL|walletType=null|holderAuthentMethod=NOT_SPECIFIED',
'Encode': '',
'InterfaceVersion': 'HP_2.4',
'Seal': '8a4c1f8b268832600a7bf40ddaa5d487f07d61dea81b8119ab8bab3c8a0861f3',
'locale': 'en',
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
addons/payment_sips/tests/test_sips.py
+81 -63
View File
@@ -1,19 +1,23 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import json
from unittest.mock import patch
from freezegun import freeze_time
from werkzeug.exceptions import Forbidden
from odoo.exceptions import ValidationError
from odoo.tests import tagged
from odoo.tools import mute_logger
from .common import SipsCommon
from ..controllers.main import SipsController
from ..models.payment_acquirer import SUPPORTED_CURRENCIES
from odoo.addons.payment.tests.http_common import PaymentHttpCommon
from odoo.addons.payment_sips.controllers.main import SipsController
from odoo.addons.payment_sips.models.payment_acquirer import SUPPORTED_CURRENCIES
from odoo.addons.payment_sips.tests.common import SipsCommon
@tagged('post_install', '-at_install')
class SipsTest(SipsCommon):
class SipsTest(SipsCommon, PaymentHttpCommon):
def test_compatible_acquirers(self):
for curr in SUPPORTED_CURRENCIES:
@@ -51,71 +55,85 @@ class SipsTest(SipsCommon):
self.assertEqual(form_info['action'], self.sips.sips_test_url)
self.assertEqual(form_inputs['InterfaceVersion'], self.sips.sips_version)
return_url = self._build_url(SipsController._return_url)
notify_url = self._build_url(SipsController._notify_url)
self.assertEqual(form_inputs['Data'],
f'amount=111111|currencyCode=978|merchantId=dummy_mid|normalReturnUrl={return_url}|' \
f'automaticResponseUrl={notify_url}|transactionReference={self.reference}|' \
f'statementReference={self.reference}|keyVersion={self.sips.sips_key_version}|' \
f'returnContext={json.dumps(dict(reference=self.reference))}'
notify_url = self._build_url(SipsController._webhook_url)
self.assertEqual(
form_inputs['Data'],
f'amount=111111|currencyCode=978|merchantId=dummy_mid|normalReturnUrl={return_url}|'
f'automaticResponseUrl={notify_url}|transactionReference={self.reference}|'
f'statementReference={self.reference}|keyVersion={self.sips.sips_key_version}|'
f'returnContext={json.dumps(dict(reference=self.reference))}',
)
self.assertEqual(
form_inputs['Seal'], '99d1d2d46a841de7fe313ac0b2d13a9e42cad50b444d35bf901879305818d9b2'
)
self.assertEqual(form_inputs['Seal'],
'4d7cc67c0168e8ce11c25fbe1937231c644861e320702ab544022b032b9eb4a2')
def test_feedback_processing(self):
# typical data posted by Sips after client has successfully paid
sips_post_data = {
'Data': 'captureDay=0|captureMode=AUTHOR_CAPTURE|currencyCode=840|'
'merchantId=002001000000001|orderChannel=INTERNET|'
'responseCode=00|transactionDateTime=2020-04-08T06:15:59+02:00|'
'transactionReference=SO100x1|keyVersion=1|'
'acquirerResponseCode=00|amount=31400|authorisationId=0020000006791167|'
'paymentMeanBrand=IDEAL|paymentMeanType=CREDIT_TRANSFER|'
'customerIpAddress=127.0.0.1|returnContext={"return_url": '
'"/payment/process", "reference": '
'"SO100x1"}|holderAuthentRelegation=N|holderAuthentStatus=|'
'transactionOrigin=INTERNET|paymentPattern=ONE_SHOT|customerMobilePhone=null|'
'mandateAuthentMethod=null|mandateUsage=null|transactionActors=null|'
'mandateId=null|captureLimitDate=20200408|dccStatus=null|dccResponseCode=null|'
'dccAmount=null|dccCurrencyCode=null|dccExchangeRate=null|'
'dccExchangeRateValidity=null|dccProvider=null|'
'statementReference=SO100x1|panEntryMode=MANUAL|walletType=null|'
'holderAuthentMethod=NO_AUTHENT_METHOD',
'Encode': '',
'InterfaceVersion': 'HP_2.4',
'Seal': 'f03f64da6f57c171904d12bf709b1d6d3385131ac914e97a7e1db075ed438f3e',
'locale': 'en',
}
# Unknown transaction
with self.assertRaises(ValidationError):
self.env['payment.transaction']._handle_feedback_data('sips', self.NOTIFICATION_DATA)
with self.assertRaises(ValidationError): # unknown transaction
self.env['payment.transaction']._handle_feedback_data('sips', sips_post_data)
# Confirmed transaction
tx = self.create_transaction('redirect')
self.env['payment.transaction']._handle_feedback_data('sips', self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
self.assertEqual(tx.acquirer_reference, self.reference)
self.amount = 314.0
self.reference = 'SO100x1'
# Cancelled transaction
old_reference = self.reference
self.reference = 'Test Transaction 2'
tx = self.create_transaction('redirect')
payload = dict(
self.NOTIFICATION_DATA,
Data=self.NOTIFICATION_DATA['Data'].replace(old_reference, self.reference)
.replace('responseCode=00', 'responseCode=12')
)
self.env['payment.transaction']._handle_feedback_data('sips', payload)
self.assertEqual(tx.state, 'cancel')
tx = self.create_transaction(flow="redirect")
@mute_logger('odoo.addons.payment_sips.controllers.main')
def test_webhook_notification_confirms_transaction(self):
""" Test the processing of a webhook notification. """
tx = self.create_transaction('redirect')
url = self._build_url(SipsController._return_url)
with patch(
'odoo.addons.payment_sips.controllers.main.SipsController'
'._verify_notification_signature'
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
# Validate the transaction
self.env['payment.transaction']._handle_feedback_data('sips', sips_post_data)
self.assertEqual(tx.state, 'done', 'Sips: validation did not put tx into done state')
self.assertEqual(tx.acquirer_reference, self.reference, 'Sips: validation did not update tx id')
@mute_logger('odoo.addons.payment_sips.controllers.main')
def test_webhook_notification_triggers_signature_check(self):
""" Test that receiving a webhook notification triggers a signature check. """
self.create_transaction('redirect')
url = self._build_url(SipsController._webhook_url)
with patch(
'odoo.addons.payment_sips.controllers.main.SipsController'
'._verify_notification_signature'
) as signature_check_mock, patch(
'odoo.addons.payment.models.payment_transaction.PaymentTransaction'
'._handle_feedback_data'
):
self._make_http_post_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(signature_check_mock.call_count, 1)
# same process for an payment in error on sips's end
sips_post_data = {
'Data': 'captureDay=0|captureMode=AUTHOR_CAPTURE|currencyCode=840|'
'merchantId=002001000000001|orderChannel=INTERNET|responseCode=12|'
'transactionDateTime=2020-04-08T06:24:08+02:00|transactionReference=SO100x2|'
'keyVersion=1|amount=31400|customerIpAddress=127.0.0.1|returnContext={"return_url": '
'"/payment/process", "reference": '
'"SO100x2"}|paymentPattern=ONE_SHOT|customerMobilePhone=null|mandateAuthentMethod=null|'
'mandateUsage=null|transactionActors=null|mandateId=null|captureLimitDate=null|'
'dccStatus=null|dccResponseCode=null|dccAmount=null|dccCurrencyCode=null|'
'dccExchangeRate=null|dccExchangeRateValidity=null|dccProvider=null|'
'statementReference=SO100x2|panEntryMode=null|walletType=null|holderAuthentMethod=null',
'InterfaceVersion': 'HP_2.4',
'Seal': '6e1995ea5432580860a04d8515b6eb1507996f97b3c5fa04fb6d9568121a16a2'
}
self.reference = 'SO100x2'
tx2 = self.create_transaction(flow="redirect")
def test_accept_notification_with_valid_signature(self):
""" Test the verification of a notification with a valid signature. """
tx = self.create_transaction('redirect')
self._assert_does_not_raise(
Forbidden, SipsController._verify_notification_signature, self.NOTIFICATION_DATA, tx
)
self.env['payment.transaction']._handle_feedback_data('sips', sips_post_data)
self.assertEqual(tx2.state, 'cancel', 'Sips: erroneous validation did not put tx into error state')
@mute_logger('odoo.addons.payment_sips.controllers.main')
def test_reject_notification_with_missing_signature(self):
""" Test the verification of a notification with a missing signature. """
tx = self.create_transaction('redirect')
payload = dict(self.NOTIFICATION_DATA, Seal=None)
self.assertRaises(Forbidden, SipsController._verify_notification_signature, payload, tx)
@mute_logger('odoo.addons.payment_sips.controllers.main')
def test_reject_notification_with_invalid_signature(self):
""" Test the verification of a notification with an invalid signature. """
tx = self.create_transaction('redirect')
payload = dict(self.NOTIFICATION_DATA, Seal='dummy')
self.assertRaises(Forbidden, SipsController._verify_notification_signature, payload, tx)
addons/payment_stripe/controllers/main.py
+72 -62
View File
@@ -7,12 +7,11 @@ import logging
import pprint
from datetime import datetime
import werkzeug
from werkzeug.exceptions import Forbidden
from odoo import http
from odoo.exceptions import ValidationError
from odoo.http import request
from odoo.tools import consteq
_logger = logging.getLogger(__name__)
@@ -20,22 +19,22 @@ _logger = logging.getLogger(__name__)
class StripeController(http.Controller):
_checkout_return_url = '/payment/stripe/checkout_return'
_validation_return_url = '/payment/stripe/validation_return'
_webhook_url = '/payment/stripe/webhook'
WEBHOOK_AGE_TOLERANCE = 10*60 # seconds
@http.route(_checkout_return_url, type='http', auth='public', csrf=False)
def stripe_return_from_checkout(self, **data):
""" Process the data returned by Stripe after redirection for checkout.
""" Process the notification data sent by Stripe after redirection from checkout.
:param dict data: The GET params appended to the URL in `_stripe_create_checkout_session`
"""
# Retrieve the tx and acquirer based on the tx reference included in the return url
# Retrieve the tx based on the tx reference included in the return url
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'stripe', data
)
acquirer_sudo = tx_sudo.acquirer_id
# Fetch the PaymentIntent, Charge and PaymentMethod objects from Stripe
payment_intent = acquirer_sudo._stripe_make_request(
payment_intent = tx_sudo.acquirer_id._stripe_make_request(
f'payment_intents/{tx_sudo.stripe_payment_intent}', method='GET'
)
_logger.info("received payment_intents response:\n%s", pprint.pformat(payment_intent))
@@ -49,17 +48,17 @@ class StripeController(http.Controller):
@http.route(_validation_return_url, type='http', auth='public', csrf=False)
def stripe_return_from_validation(self, **data):
""" Process the data returned by Stripe after redirection for validation.
""" Process the notification data sent by Stripe after redirection for validation.
:param dict data: The GET params appended to the URL in `_stripe_create_checkout_session`
"""
# Retrieve the acquirer based on the tx reference included in the return url
acquirer_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
# Retrieve the transaction based on the tx reference included in the return url
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'stripe', data
).acquirer_id
)
# Fetch the Session, SetupIntent and PaymentMethod objects from Stripe
checkout_session = acquirer_sudo._stripe_make_request(
checkout_session = tx_sudo.acquirer_id._stripe_make_request(
f'checkout/sessions/{data.get("checkout_session_id")}',
payload={'expand[]': 'setup_intent.payment_method'}, # Expand all required objects
method='GET'
@@ -73,49 +72,51 @@ class StripeController(http.Controller):
# Redirect the user to the status page
return request.redirect('/payment/status')
@http.route('/payment/stripe/webhook', type='json', auth='public')
@http.route(_webhook_url, type='json', auth='public')
def stripe_webhook(self):
""" Process the `checkout.session.completed` event sent by Stripe to the webhook.
""" Process the notification data sent by Stripe to the webhook.
:return: An empty string to acknowledge the notification with an HTTP 200 response
:return: An empty string to acknowledge the notification
:rtype: str
"""
event = json.loads(request.httprequest.data)
_logger.info("event received:\n%s", pprint.pformat(event))
_logger.info("notification received from Stripe with data:\n%s", pprint.pformat(event))
try:
if event['type'] == 'checkout.session.completed':
checkout_session = event['data']['object']
# Check the source and integrity of the event
# Check the integrity of the event
data = {'reference': checkout_session['client_reference_id']}
tx_sudo = request.env['payment.transaction'].sudo()._get_tx_from_feedback_data(
'stripe', data
)
if self._verify_webhook_signature(tx_sudo.acquirer_id.stripe_webhook_secret):
# Fetch the PaymentIntent, Charge and PaymentMethod objects from Stripe
if checkout_session.get('payment_intent'): # Can be None
payment_intent = tx_sudo.acquirer_id._stripe_make_request(
f'payment_intents/{tx_sudo.stripe_payment_intent}', method='GET'
)
_logger.info(
"received payment_intents response:\n%s", pprint.pformat(payment_intent)
)
self._include_payment_intent_in_feedback_data(payment_intent, data)
# Fetch the SetupIntent and PaymentMethod objects from Stripe
if checkout_session.get('setup_intent'): # Can be None
setup_intent = tx_sudo.acquirer_id._stripe_make_request(
f'setup_intents/{checkout_session.get("setup_intent")}',
payload={'expand[]': 'payment_method'},
method='GET'
)
_logger.info(
"received setup_intents response:\n%s", pprint.pformat(setup_intent)
)
self._include_setup_intent_in_feedback_data(setup_intent, data)
# Handle the feedback data crafted with Stripe API objects as a regular feedback
request.env['payment.transaction'].sudo()._handle_feedback_data('stripe', data)
self._verify_notification_signature(tx_sudo)
# Fetch the PaymentIntent, Charge and PaymentMethod objects from Stripe
if checkout_session.get('payment_intent'): # Can be None
payment_intent = tx_sudo.acquirer_id._stripe_make_request(
f'payment_intents/{tx_sudo.stripe_payment_intent}', method='GET'
)
_logger.info(
"received payment_intents response:\n%s", pprint.pformat(payment_intent)
)
self._include_payment_intent_in_feedback_data(payment_intent, data)
# Fetch the SetupIntent and PaymentMethod objects from Stripe
if checkout_session.get('setup_intent'): # Can be None
setup_intent = tx_sudo.acquirer_id._stripe_make_request(
f'setup_intents/{checkout_session.get("setup_intent")}',
payload={'expand[]': 'payment_method'},
method='GET'
)
_logger.info(
"received setup_intents response:\n%s", pprint.pformat(setup_intent)
)
self._include_setup_intent_in_feedback_data(setup_intent, data)
# Handle the feedback data crafted with Stripe API objects as a regular feedback
request.env['payment.transaction'].sudo()._handle_feedback_data('stripe', data)
except ValidationError: # Acknowledge the notification to avoid getting spammed
_logger.exception("unable to handle the data; skipping to acknowledge the notification")
_logger.exception("unable to handle the notification data; skipping to acknowledge")
return ''
@staticmethod
@@ -135,39 +136,48 @@ class StripeController(http.Controller):
'payment_method': setup_intent.get('payment_method')
})
def _verify_webhook_signature(self, webhook_secret):
""" Check that the signature computed from the feedback matches the received one.
def _verify_notification_signature(self, tx_sudo):
""" Check that the received signature matches the expected one.
See https://stripe.com/docs/webhooks/signatures#verify-manually.
:param str webhook_secret: The secret webhook key of the acquirer handling the transaction
:return: Whether the signatures match
:rtype: str
:param recordset tx_sudo: The sudoed transaction referenced by the notification data, as a
`payment.transaction` record
:return: None
:raise: :class:`werkzeug.exceptions.Forbidden` if the timestamp is too old or if the
signatures don't match
"""
webhook_secret = tx_sudo.acquirer_id.stripe_webhook_secret
if not webhook_secret:
_logger.warning("ignored webhook event due to undefined webhook secret")
return False
return
notification_payload = request.httprequest.data.decode('utf-8')
signature_entries = request.httprequest.headers.get('Stripe-Signature').split(',')
signature_entries = request.httprequest.headers['Stripe-Signature'].split(',')
signature_data = {k: v for k, v in [entry.split('=') for entry in signature_entries]}
# Check the timestamp of the event
event_timestamp = int(signature_data['t'])
if datetime.utcnow().timestamp() - event_timestamp > self.WEBHOOK_AGE_TOLERANCE:
_logger.warning("ignored webhook event due to age tolerance: %s", event_timestamp)
return False
# Retrieve the timestamp from the data
event_timestamp = int(signature_data.get('t', '0'))
if not event_timestamp:
_logger.warning("received notification with missing timestamp")
raise Forbidden()
# Compare signatures
received_signature = signature_data['v1']
# Check if the timestamp is not too old
if datetime.utcnow().timestamp() - event_timestamp > self.WEBHOOK_AGE_TOLERANCE:
_logger.warning("received notification with outdated timestamp: %s", event_timestamp)
raise Forbidden()
# Retrieve the received signature from the data
received_signature = signature_data.get('v1')
if not received_signature:
_logger.warning("received notification with missing signature")
raise Forbidden()
# Compare the received signature with the expected signature computed from the data
signed_payload = f'{event_timestamp}.{notification_payload}'
expected_signature = hmac.new(
webhook_secret.encode('utf-8'),
signed_payload.encode('utf-8'),
hashlib.sha256
webhook_secret.encode('utf-8'), signed_payload.encode('utf-8'), hashlib.sha256
).hexdigest()
if not consteq(received_signature, expected_signature):
_logger.warning("ignored event with invalid signature")
return False
return True
if not hmac.compare_digest(received_signature, expected_signature):
_logger.warning("received notification with invalid signature")
raise Forbidden()
addons/payment_stripe/tests/common.py
+57
View File
@@ -4,6 +4,63 @@ from odoo.addons.payment.tests.common import PaymentCommon
class StripeCommon(PaymentCommon):
NOTIFICATION_DATA = {
'api_version': '2019-05-16',
'created': 1326853478,
'data': {
'object': {
'after_expiration': None,
'allow_promotion_codes': None,
'amount_subtotal': None,
'amount_total': None,
'automatic_tax': {
'enabled': False,
'status': None,
},
'billing_address_collection': None,
'cancel_url': 'https://example.com/payment/stripe/checkout_return?reference=Test Transaction',
'client_reference_id': 'Test Transaction', # Shamefully copy-pasted from payment
'consent': None,
'consent_collection': None,
'currency': None,
'customer': None,
'customer_creation': None,
'customer_details': None,
'customer_email': None,
'expires_at': 1642614393,
'id': 'cs_00000000000000',
'livemode': False,
'locale': None,
'metadata': {},
'mode': 'payment',
'object': 'checkout.session',
'payment_intent': 'pi_00000000000000',
'payment_method_options': {},
'payment_method_types': ['card'],
'payment_status': 'unpaid',
'phone_number_collection': {'enabled': False},
'recovered_from': None,
'setup_intent': None,
'shipping': None,
'shipping_address_collection': None,
'shipping_options': [],
'shipping_rate': None,
'status': 'expired',
'submit_type': None,
'subscription': None,
'success_url': 'https://example.com/payment/stripe/checkout_return?reference=Test Transaction',
'total_details': None,
'url': None,
},
},
'id': 'evt_00000000000000',
'livemode': False,
'object': 'event',
'pending_webhooks': 1,
'request': None,
'type': 'checkout.session.completed',
}
@classmethod
def setUpClass(cls, chart_template_ref=None):
super().setUpClass(chart_template_ref=chart_template_ref)
addons/payment_stripe/tests/test_stripe.py
+42 -5
View File
@@ -5,25 +5,62 @@ from unittest.mock import patch
from odoo.tests import tagged
from odoo.tools import mute_logger
from .common import StripeCommon
from odoo.addons.payment.tests.http_common import PaymentHttpCommon
from odoo.addons.payment_stripe.controllers.main import StripeController
from odoo.addons.payment_stripe.tests.common import StripeCommon
@tagged('post_install', '-at_install')
class StripeTest(StripeCommon):
class StripeTest(StripeCommon, PaymentHttpCommon):
def test_processing_values(self):
dummy_session_id = 'cs_test_sbTG0yGwTszAqFUP8Ulecr1bUwEyQEo29M8taYvdP7UA6Qr37qX6uA6w'
tx = self.create_transaction(flow='redirect') # We don't really care what the flow is here.
tx = self.create_transaction(flow='redirect') # We don't really care what the flow is here.
# Ensure no external API call is done, we only want to check the processing values logic
def mock_stripe_create_checkout_session(self):
return {'id': dummy_session_id}
with patch.object(
type(self.env['payment.transaction']),
'_stripe_create_checkout_session',
type(self.env['payment.transaction']), '_stripe_create_checkout_session',
mock_stripe_create_checkout_session,
), mute_logger('odoo.addons.payment.models.payment_transaction'):
processing_values = tx._get_processing_values()
self.assertEqual(processing_values['publishable_key'], self.stripe.stripe_publishable_key)
self.assertEqual(processing_values['session_id'], dummy_session_id)
@mute_logger('odoo.addons.payment_stripe.controllers.main')
def test_webhook_notification_confirms_transaction(self):
""" Test the processing of a webhook notification. """
tx = self.create_transaction('redirect')
url = self._build_url(StripeController._webhook_url)
with patch(
'odoo.addons.payment_stripe.controllers.main.StripeController'
'._verify_notification_signature'
), patch(
'odoo.addons.payment_stripe.models.payment_acquirer.PaymentAcquirer'
'._stripe_make_request',
return_value={'status': 'succeeded'},
):
self._make_json_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(tx.state, 'done')
@mute_logger('odoo.addons.payment_stripe.controllers.main')
def test_webhook_notification_triggers_signature_check(self):
""" Test that receiving a webhook notification triggers a signature check. """
self.create_transaction('redirect')
url = self._build_url(StripeController._webhook_url)
with patch(
'odoo.addons.payment_stripe.controllers.main.StripeController'
'._verify_notification_signature'
) as signature_check_mock, patch(
'odoo.addons.payment_stripe.models.payment_acquirer.PaymentAcquirer'
'._stripe_make_request',
return_value={},
), patch(
'odoo.addons.payment.models.payment_transaction.PaymentTransaction'
'._handle_feedback_data'
):
self._make_json_request(url, data=self.NOTIFICATION_DATA)
self.assertEqual(signature_check_mock.call_count, 1)