[Python] Square Webhook doesn't give confirmation if the payment is done or not to...

I'm encountering a critical issue where my Django backend is unable to receive confirmation from Square about whether a payment has been completed.

Here are the environment details for reference:

• Python: 3.13.2
• Django: 5.1.6
• Square SDK: 43.0.0.20250618

Problem Summary:

The core flow of the payment process is as follows:

• The user creates an order via the CreateSquareOrder method.
• This successfully generates a Square-hosted checkout page where the user completes the payment.
• The payment is processed correctly—I can confirm the funds are reaching my Square account.

However, the issue arises after payment is completed:

• The webhook I’ve configured (https://api.mydomain.com/api/wallet/webhooks/square/) is supposed to be triggered automatically by Square to confirm the payment.
• Unfortunately, this webhook is not being triggered properly. I’ve checked the server logs, and I consistently receive a 401 Unauthorized response from Square's webhook call attempts.
• Because the webhook is failing, the subsequent logic in my Django app—specifically, the VerifySquareOrder method—is not executed, meaning critical application records are never created.

Additional Information:

Below is the backend code I’m currently using to handle CreateSquareOrder, SquareWebhookView and VerifySquareOrder requests. I’ve followed the latest version guidelines and best practices as per the Square documentation:

import uuid
from decimal import Decimal
from django.utils import timezone
from django.conf import settings
from dateutil.relativedelta import relativedelta
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, AllowAny
from rest_framework import status
from square import Square
from square.environment import SquareEnvironment
import hmac
import hashlib
import base64
from django.views.decorators.csrf import csrf_exempt
from django.utils.decorators import method_decorator
from rest_framework.request import Request
from rest_framework.parsers import JSONParser
from users.models import CustomUser
from subscription.models import Subscription
from wallet.models import Wallet, Transaction, SecurityDeposit

DOMAIN = settings.DOMAIN
SQUARE_ACCESS_TOKEN = settings.SQUARE_ACCESS_TOKEN
SQUARE_LOCATION_ID = settings.SQUARE_LOCATION_ID
SQUARE_ENVIRONMENT = SquareEnvironment.PRODUCTION
SQUARE_WEBHOOK_SIGNATURE_KEY = settings.SQUARE_WEBHOOK_SIGNATURE_KEY

class CreateSquareOrder(APIView):
permission_classes = [IsAuthenticated]

def post(self, request):
user = request.user
amount = request.data.get("amount")
plan = request.data.get("plan")

if not amount or not plan:
return Response({"error": "Amount and plan are required"}, status=status.HTTP_400_BAD_REQUEST)

try:
client = Square(token=SQUARE_ACCESS_TOKEN, environment=SQUARE_ENVIRONMENT)
idempotency_key = str(uuid.uuid4())

response = client.checkout.payment_links.create(
idempotency_key=idempotency_key,
order={
"location_id": SQUARE_LOCATION_ID,
"line_items": [
{
"name": f"Plan {plan}",
"quantity": "1",
"base_price_money": {
"amount": int(float(amount) * 100),
"currency": "AUD"
}
}
],
"metadata": {
"user_id": str(user.id),
"plan": str(plan),
"amount": str(amount),
"purpose": "subscription"
}
},
checkout_options={
"redirect_url": f"{DOMAIN}/square-success",
"cancel_url": f"{DOMAIN}/square-cancel"
}
)

if response.errors:
return Response({
"error": "Failed to create Square checkout",
"details": response.errors
}, status=400)

return Response({"checkout_url": response.payment_link.url})

except Exception as e:
return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

@method_decorator(csrf_exempt, name='dispatch')
class SquareWebhookView(APIView):
permission_classes = []
def post(self, request):
signature = request.headers.get("X-Square-Signature", "")
request_body = request.body.decode()

if not is_valid_square_signature(signature, request_body, SQUARE_WEBHOOK_SIGNATURE_KEY):
return Response({"error": "Invalid signature"}, status=401)

event_type = request.data.get("type", "")
data = request.data.get("data", {}).get("object", {})
payment = data.get("payment", {})
payment_id = payment.get("id")
status_ = payment.get("status")

if event_type == "payment.updated" and status_ == "COMPLETED":
try:
new_request = Request(request._request, data={'payment_id': payment_id}, parsers=[JSONParser()])
return VerifySquareOrder().post(new_request)
except Exception as e:
print(f"Error processing payment webhook: {e}")
return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

return Response(status=200)

def is_valid_square_signature(signature, request_body, signature_key):
key_bytes = bytes(signature_key, 'utf-8')
body_bytes = bytes(request_body, 'utf-8')
hmac_hash = hmac.new(key_bytes, body_bytes, hashlib.sha256).digest()
computed_signature = base64.b64encode(hmac_hash).decode()
return hmac.compare_digest(signature, computed_signature)

class VerifySquareOrder(APIView):
permission_classes = [AllowAny]

def post(self, request):
try:
payment_id = request.data.get("payment_id")
if not payment_id:
return Response({"error": "Payment ID is required"}, status=400)

client = Square(token=SQUARE_ACCESS_TOKEN, environment=SQUARE_ENVIRONMENT)
payment_response = client.payments.get_payment(payment_id=payment_id)

if payment_response.errors:
print(f"Square Payment API Errors: {payment_response.errors}")
return Response({"error": "Failed to retrieve Square payment", "details": payment_response.errors}, status=400)

payment_data = payment_response.payment
if payment_data.status != "COMPLETED":
return Response({"error": f"Payment not completed. Current status: {payment_data.status}"}, status=400)

order_id = payment_data.order_id
if not order_id:
return Response({"error": "Order ID not found for this payment."}, status=400)

order_response = client.orders.retrieve_order(order_id=order_id, location_id=SQUARE_LOCATION_ID)
if order_response.errors:
print(f"Square Order API Errors: {order_response.errors}")
return Response({"error": "Failed to retrieve Square order associated with payment", "details": order_response.errors}, status=400)

order_data = order_response.order
metadata = order_data.metadata

if not metadata:
return Response({"error": "Order metadata not found."}, status=400)

user = CustomUser.objects.get(id=metadata["user_id"])
wallet, _ = Wallet.objects.get_or_create(user=user)

amount = Decimal(metadata["amount"])
plan = int(metadata["plan"])
description = f"Square - {metadata.get('purpose', 'subscription')} [order_id:{order_id}, payment_id:{payment_id}]"

# Here i will create records that i need in my website

return Response({"message": "Payment verified and subscription activated."})

except CustomUser.DoesNotExist:
print("User not found based on metadata user_id.")
return Response({"error": "User specified in Square order metadata not found."}, status=404)
except Exception as e:
print(f"Error verifying Square order: {e}")
return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)


What I Need:

• Can you please verify why Square is returning a 401 error for the webhook URL?
• Is there any additional authentication or header setup required that might be missing?
• I need to resolve this urgently, as this issue is blocking the payment verification and data sync process for all users.

I’d appreciate your assistance and suggestions. Thank you folks

Continue reading...