Checksum Marking Documentation

Overview

This document explains how the notify function marks checksum verification status in the CardPaymentRecordTable. The checksum verification process ensures that webhook notifications from KBank payment gateway are authentic and have not been tampered with.

Function Flow

Entry Point

File: functions/callback/notify.py
Handler: lambda_handler(event, _)

The checksum verification happens at the very beginning of the notify function execution:

def lambda_handler(event, _):
    """Record credit card callback."""
    # Reset delayed notify flag for this invocation
    reset_delayed_notify_flag()
    
    processor = None
    try:
        # Verify checksum FIRST before any other processing
        checksum_valid, checksum_error, charge_id = verify_and_handle_checksum(
            event=event,
            function_name="notify",
            branch=BRANCH,
            debug_mode=DEBUG_MODE
        )

Checksum Verification Process

File: functions/callback/src/checksum_handler.py
Function: verify_and_handle_checksum()

The process follows these steps:

1. Extract charge_id from event body
2. Verify checksum using KBank secret key
3. Update checksumVerified field in CardPaymentRecordTable
4. Handle errors if checksum fails (logging, email notifications, Sentry)

def verify_and_handle_checksum(
    event: dict,
    function_name: str,
    branch: Optional[str] = None,
    debug_mode: bool = False
) -> Tuple[bool, Optional[str], Optional[str]]:
    """
    Verify checksum and handle status updates and error logging.
    
    This function:
    1. Extracts charge_id from event
    2. Verifies checksum
    3. Updates checksumVerified status in CardPaymentRecordTable
    4. Handles error logging and email notifications if checksum fails
    
    Args:
        event: Lambda event containing 'body' field
        function_name: Name of the calling function (for logging)
        branch: Branch name for secret lookup (optional)
        debug_mode: If True, skip database updates
        
    Returns:
        Tuple of (is_valid, error_message, charge_id)
        - is_valid: True if checksum is valid, False otherwise
        - error_message: Error message if checksum failed, None otherwise
        - charge_id: Extracted charge ID or None
    """
    # Extract charge_id from event body
    charge_id = extract_charge_id(event)
    
    # Verify checksum
    checksum_valid, checksum_error = verify_notification_checksum(event, branch=branch, debug=debug_mode)
    
    if not checksum_valid:
        error_msg = f"Checksum verification failed: {checksum_error}"
        print(f"[ERROR] {error_msg}")
        
        # Update checksum verification status to False
        if charge_id:
            update_checksum_verification_status(charge_id, False, debug_mode)
        
        # Log checksum verification failure
        checksum_exception = Exception(error_msg)
        log_callback_error(
            error=checksum_exception,
            function_name=function_name,
            event=event,
            orderId=None,
            chargeId=charge_id,
            response_data=None
        )
        send_callback_error_email(
            error=checksum_exception,
            function_name=function_name,
            orderId=None,
            chargeId=charge_id,
            event=event
        )
        sentry_sdk.capture_exception(checksum_exception)
        
        return False, error_msg, charge_id
    
    print("[SUCCESS] Checksum verification passed")
    
    # Update checksum verification status to True
    if charge_id:
        update_checksum_verification_status(charge_id, True, debug_mode)
    
    return True, None, charge_id

DynamoDB Table Schema

CardPaymentRecordTable (Branch Table)

Table Name: payment3-card-payment-record-{BRANCH}
Region: ap-southeast-1

PynamoDB Model:

class CardPaymentRecordTable(Model):
    class Meta:
        table_name = f"payment3-card-payment-record-{BRANCH}"
        region = "ap-southeast-1"

    chargeId_index = ChargeIdIndex()
    tokenId = UnicodeAttribute(hash_key=True)
    orderId = UnicodeAttribute()
    createdAt = UTCDateTimeAttribute()
    chargeId = UnicodeAttribute(null=True)
    status = BooleanAttribute(null=True)
    checksumVerified = BooleanAttribute(null=True)

    def __post_init__(self):
        self.createdAt = datetime.now()

Key Fields:

• Hash Key: tokenId (UnicodeAttribute)
• Global Secondary Index: chargeId_index (hash_key: chargeId)
• checksumVerified: BooleanAttribute(null=True) - Marks if checksum was verified
• status: BooleanAttribute(null=True) - Payment status (True = success, False = failure)

CardPaymentRecordTableMaster (Master Table)

Table Name: payment3-card-payment-record-master
Region: ap-southeast-1

PynamoDB Model:

class CardPaymentRecordTableMaster(Model):
    class Meta:
        table_name = f"payment3-card-payment-record-master"
        region = "ap-southeast-1"

    chargeId_index = ChargeIdIndex()
    tokenId = UnicodeAttribute(hash_key=True)
    orderId = UnicodeAttribute()
    createdAt = UTCDateTimeAttribute()
    chargeId = UnicodeAttribute(null=True)
    status = BooleanAttribute(null=True)
    checksumVerified = BooleanAttribute(null=True)

    def __post_init__(self):
        self.createdAt = datetime.now()

Note: Master table has the same schema as branch table. Used as fallback when record is not found in branch table.

Updating Checksum Verification Status

Function: update_checksum_verification_status()

File: functions/callback/src/checksum_handler.py

This function updates the checksumVerified field in the CardPaymentRecordTable:

def update_checksum_verification_status(charge_id: str, verified: bool, debug_mode: bool = False):
    """
    Update checksumVerified field in CardPaymentRecordTable.
    
    Args:
        charge_id: The charge ID to update
        verified: Boolean indicating if checksum was verified successfully
        debug_mode: If True, skip actual database updates
    """
    if debug_mode:
        print(f"DEBUG_MODE: Skipping checksum verification status update. Would update chargeId={charge_id}, checksumVerified={verified}")
        return
    
    try:
        # Try branch-specific table first
        try:
            query_result = CardPaymentRecordTable.chargeId_index.query(charge_id)
            record = next(query_result)
            record.checksumVerified = verified
            record.save()
            print(f"[SUCCESS] Updated checksumVerified={verified} for chargeId={charge_id} in branch table")
            return
        except StopIteration:
            print(f"[INFO] Record not found in branch table, trying master table")
        except Exception as e:
            print(f"[WARNING] Error updating branch table: {type(e).__name__}: {e}")
        
        # Try master table
        try:
            query_result = CardPaymentRecordTableMaster.chargeId_index.query(charge_id)
            record = next(query_result)
            record.checksumVerified = verified
            record.save()
            print(f"[SUCCESS] Updated checksumVerified={verified} for chargeId={charge_id} in master table")
            return
        except StopIteration:
            print(f"[WARNING] Card payment record not found in both tables for chargeId={charge_id}")
        except Exception as e:
            print(f"[WARNING] Error updating master table: {type(e).__name__}: {e}")
    except Exception as e:
        print(f"[ERROR] Failed to update checksum verification status: {type(e).__name__}: {e}")
        # Don't raise - this is a non-critical update

Query Pattern

The function uses the chargeId_index Global Secondary Index to find records:

PynamoDB Query Example:

from src.cardPaymentRecordTable import CardPaymentRecordTable

# Query by chargeId using GSI
charge_id = "chrg_prod_1234567890"
query_result = CardPaymentRecordTable.chargeId_index.query(charge_id)

try:
    record = next(query_result)
    # Update checksumVerified field
    record.checksumVerified = True  # or False
    record.save()
    print(f"Updated checksumVerified for chargeId={charge_id}")
except StopIteration:
    print(f"Record not found for chargeId={charge_id}")

Fallback Logic

The update process follows this fallback pattern:

1. First: Try to find record in branch-specific table (payment3-card-payment-record-{BRANCH})
2. If not found: Try master table (payment3-card-payment-record-master)
3. If still not found: Log warning but don't raise exception (non-critical update)

This fallback ensures that records created in master branch can still have their checksum status updated.

Charge ID Extraction

Function: extract_charge_id()

def extract_charge_id(event: dict) -> Optional[str]:
    """
    Extract charge ID from event body.
    
    Args:
        event: Lambda event containing 'body' field
        
    Returns:
        Charge ID string or None if extraction fails
    """
    try:
        if isinstance(event.get("body"), str):
            body_data = json.loads(event["body"])
        else:
            body_data = event.get("body", {})
        return body_data.get("objectId") or body_data.get("id")
    except Exception as e:
        print(f"[WARNING] Could not extract charge_id from event: {e}")
        return None

Charge ID Sources:

• Primary: body.objectId (from KBank callback)
• Fallback: body.id (alternative field name)

Checksum Verification Values

Possible Values for checksumVerified

• True: Checksum was successfully verified
• False: Checksum verification failed
• None: Checksum verification was not attempted or charge_id was not found

When Checksum is Marked

1. True: When verify_notification_checksum() returns True
2. False: When verify_notification_checksum() returns False or raises an exception
3. None: When charge_id cannot be extracted from event

Usage in Other Functions

The checksumVerified field can be used by other functions (like processPaidOrder) to determine if they can trust the payment status from CardPaymentRecordTable.status without needing to query CheckStatusTable.

Example Usage:

from src.cardPaymentRecordTable import CardPaymentRecordTable

# Query card payment record by orderId
order_id = "492500005643"
query_result = CardPaymentRecordTable.orderId_index.query(order_id)

try:
    record = next(query_result)
    
    # Check if checksum is verified
    if record.checksumVerified is True:
        # Trust the status field directly
        payment_success = record.status is True
        print(f"Payment success (from verified checksum): {payment_success}")
    else:
        # Need to verify status through CheckStatusTable
        print("Checksum not verified, need to check CheckStatusTable")
        
except StopIteration:
    print(f"Card payment record not found for orderId={order_id}")

Error Handling

If checksum verification fails:

1. Status Update: checksumVerified is set to False
2. Error Logging: Error is logged to CallbackErrorLogTable
3. Email Notification: Error email is sent via SES
4. Sentry: Exception is captured in Sentry
5. Processing Continues: The function continues processing despite checksum failure (logs warning)

Related Files

• functions/callback/notify.py - Main entry point
• functions/callback/src/checksum_handler.py - Checksum verification and status update logic
• functions/callback/src/checksum_verifier.py - KBank checksum verification algorithm
• functions/callback/src/cardPaymentRecordTable.py - Branch table model
• functions/callback/src/cardPaymentRecordTableMaster.py - Master table model

Template.yaml Configuration

Function: Notify (lines 348-413 in template.yaml)

Required IAM Policies:

• DynamoDBCrudPolicy for CardPaymentRecordTable (line 372-373)
• DynamoDBCrudPolicy for payment3-card-payment-record-master (line 375-376)
• AWSSecretsManagerGetSecretValuePolicy for kbank-dev and kbank-prod secrets (lines 396-399)

These permissions are required for:

• Querying records by chargeId using GSI
• Updating checksumVerified field
• Accessing KBank secret keys for checksum verification