Python Signing And Verification

import base64
import hashlib
import logging
import re
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa

logger = logging.getLogger("payment-signature")


def _timestamp(value):
    text = str(value)
    if not re.fullmatch(r"[1-9][0-9]{0,18}", text) or int(text) > 9223372036854775807:
        raise ValueError("invalid Timestamp: use Unix seconds")
    return text


def _base64(text, label):
    try:
        result = base64.b64decode(text, validate=True)
        if not result or base64.b64encode(result).decode("ascii") != text:
            raise ValueError()
        return result
    except (ValueError, TypeError):
        raise ValueError(label + ": invalid Base64") from None


def _key(text, private):
    if not isinstance(text, str) or len(text) > 16384:
        raise ValueError("key: missing or too large")
    text = text.replace(r"\r\n", "\n").replace(r"\n", "\n").strip()
    if text.startswith("-----BEGIN"):
        match = re.fullmatch(r"-----BEGIN ([A-Z ]+)-----\s*([A-Za-z0-9+/=\s]+)\s*-----END \1-----", text)
        labels = ("PRIVATE KEY", "RSA PRIVATE KEY") if private else ("PUBLIC KEY", "RSA PUBLIC KEY")
        if not match or match[1] not in labels:
            raise ValueError("key: expected one unencrypted RSA PEM block of the correct kind")
        text = match[2]
    der = _base64("".join(text.split()), "key")
    try:
        key = serialization.load_der_private_key(der, password=None) if private else serialization.load_der_public_key(der)
        expected = rsa.RSAPrivateKey if private else rsa.RSAPublicKey
        if not isinstance(key, expected):
            raise ValueError()
        return key
    except (ValueError, TypeError):
        raise ValueError("key: invalid/truncated, encrypted or non-RSA key") from None


def _hash(body):
    if not isinstance(body, bytes):
        raise ValueError("body: expected raw bytes, not parsed JSON")
    return hashlib.sha512(body).hexdigest()


def _log(operation, body, error):
    logger.error("[payment-%s] FAILED body_bytes=%d error=%s", operation,
                 len(body) if isinstance(body, bytes) else 0, str(error))


def sign_request(path, body, timestamp, merchant_private_key):
    """None means no body; b'' means a present, empty body. Returns Base64 or raises."""
    try:
        if not isinstance(path, str) or not path.startswith("/"):
            raise ValueError("path: expected API path")
        origin = path + ("" if body is None else _hash(body)) + _timestamp(timestamp)
        key = _key(merchant_private_key, True)
        signature = key.sign(origin.encode("utf-8"), padding.PKCS1v15(), hashes.SHA512())
        return base64.b64encode(signature).decode("ascii")
    except Exception as error:
        _log("sign", body, error)
        raise


def verify_platform(body, timestamp, signature, platform_public_key):
    """Exact callback bytes, or exact signed data bytes for API envelopes. NO path."""
    try:
        origin = _hash(body) + _timestamp(timestamp)
        key = _key(platform_public_key, False)
        sig = _base64(signature, "Signature")
        expected = (key.key_size + 7) // 8
        if len(sig) != expected:
            raise ValueError(f"Signature length: got {len(sig)}, expected {expected} bytes")
        try:
            key.verify(sig, origin.encode("utf-8"), padding.PKCS1v15(), hashes.SHA512())
        except InvalidSignature:
            raise ValueError("RSA SHA512 verification failed: check platform key, raw bytes, timestamp and formula") from None
    except Exception as error:
        _log("verify", body, error)
        raise

"""Offline callback verification vector. Test public key only; never use in production."""
import base64
import hashlib

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa


PLATFORM_PUBLIC_KEY = b"""-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAskopsusffzyPW77qfwfp
/daafyzflakpa3YoGRS+RrjDyNSsJnszOSPgyjsJvaHhh3qBd4qsH4N0+G15mP8c
9s8dhz0VWySwu+Xxo+pJBjurnkGpDBclTgecL5nKDj3SwB6gJ7LR5i0Ey+DGURvL
MUEVJO2UW6ZzcpPKYpbJGsNiQa1XsB9k/Wx22MEZlbAuFu2C4dOLJkApjAOULcnF
Ik80/yFvXIvqZIfLNwtffPlkXXw/I2sxP13Gne0edQvPOAJCyUV6UmDKNILLS8X0
nNi6ZzaVSbfe0BKB/0Yq5pUktn00Ridt86oktSG7wmUmsUFL11WnGpEVN2q3m8Bv
vwIDAQAB
-----END PUBLIC KEY-----
"""
RAW_BODY = (
    b'{"transactionId":"TEST-PKR-001","merchantSerialNo":"TEST-MERCHANT-001",'
    b'"currency":"pkr","method":"BANK","amount":"100.00",'
    b'"orderAmount":"100.00","status":"20","fee":"1.00"}'
)
TIMESTAMP = "1792132760"
SIGNATURE = "XfXOGFZPUgEckD3m1+ncrXC0KiXIHxpoTjls16ULOO5gByIMCjn4LLB/+85kNzEUbq2vfnj7NRIbS/vxoG6y/N/ht/xVP89HkwWaVGYJPF5R4g46VnGatRtHSPfENm8sJXTyyqNUnHBecSxVF4k5JRlItpue8IGKmtfpzYlqlaUylDSuhAHo7ZX7JVGihSJmZxN7fHy8d+KiDV8tF3cKswvvq9aefbSWq8nG9Lw+536dey5ZV9YUxfaI+OxrG9+N0F2jR7zvKI/EiilIrO5mLNd2aN7hrdJbXY6PQOrWiY9m0+DRMQxURnGGyupLPYuFbr6ApP3jb6VfMaoUS1n7tw=="
EXPECTED_BODY_SHA512 = "e246541235790a61cd2d10a2748e817aaefb0b147c25456420f380859a7a4e43e491f67b8bb234cd0a5e08d1dda00c7cb66c66d86fae3ec23d3ba4f2627891b6"


def verify_callback(raw_body, timestamp, signature, platform_public_key):
    """Verify received raw bytes; no path, JSON sorting or re-serialization."""
    key = serialization.load_pem_public_key(platform_public_key)
    if not isinstance(key, rsa.RSAPublicKey):
        raise ValueError("Platform public key must be RSA")
    signature_bytes = base64.b64decode(signature, validate=True)
    expected_bytes = (key.key_size + 7) // 8
    if len(signature_bytes) != expected_bytes:
        raise ValueError(
            f"Signature is {len(signature_bytes)} bytes; "
            f"the {key.key_size}-bit key expects {expected_bytes} bytes"
        )
    body_hash = hashlib.sha512(raw_body).hexdigest()
    signed_text = (body_hash + timestamp).encode("utf-8")
    key.verify(signature_bytes, signed_text, padding.PKCS1v15(), hashes.SHA512())


def main():
    assert hashlib.sha512(RAW_BODY).hexdigest() == EXPECTED_BODY_SHA512
    verify_callback(RAW_BODY, TIMESTAMP, SIGNATURE, PLATFORM_PUBLIC_KEY)
    print("RSA key: 2048 bits")
    print("Signature: 256 bytes")
    print("Callback signature: PASS")

    try:
        verify_callback(RAW_BODY + b" ", TIMESTAMP, SIGNATURE, PLATFORM_PUBLIC_KEY)
    except InvalidSignature:
        print("Changed raw body: rejected (expected)")
    else:
        raise AssertionError("Changed body was accepted")


if __name__ == "__main__":
    main()

import time
import requests
from sign import sign_request


def send(base_url, path, merchant_code, raw_body, merchant_private_key):
    """raw_body is bytes or None. Never use requests' json= after signing."""
    if not base_url.startswith("https://"):
        raise ValueError("HTTPS is required")
    timestamp = str(int(time.time()))
    signature = sign_request(path, raw_body, timestamp, merchant_private_key)
    headers = {"Content-Type": "application/json", "Merchant-Code": merchant_code,
               "Timestamp": timestamp, "Signature": signature}
    with requests.request("GET" if raw_body is None else "POST", base_url + path,
                          data=raw_body, headers=headers, timeout=(10, 60),
                          allow_redirects=False, verify=True, stream=True) as response:
        if not 200 <= response.status_code < 300:
            raise RuntimeError(f"HTTP {response.status_code}; reconcile before retrying")
        chunks, size = [], 0
        for chunk in response.iter_content(65536):
            size += len(chunk)
            if size > 8 * 1024 * 1024:
                raise RuntimeError("Response exceeds 8 MiB")
            chunks.append(chunk)
        return b"".join(chunks)  # Preserve raw response for envelope verification.

import json
from http_util import send
from sign import verify_platform

body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
response_bytes = send(base_url, "/v1/merchant/pay", merchant_code, body, merchant_private_key)
# GET: send(base_url, "/v1/merchant/account", merchant_code, None, merchant_private_key)
# Verify the signed response data before using it. Do not treat HTTP 200 as payment success.

# Flask example: app.config["MAX_CONTENT_LENGTH"] = 1 << 20
@app.post("/callback")
def callback():
    raw_body = request.get_data(cache=True)
    try:
        verify_platform(raw_body, request.headers.get("Timestamp"),
                        request.headers.get("Signature"), configured_platform_public_key)
    except Exception:
        return "invalid signature", 401
    event = json.loads(raw_body)
    validate_and_apply_idempotently(event)  # Your durable business transaction.
    return "SUCCESS", 200