API Security with encrypted value in endpoints

งานนี้จะแชร์การใช้งาน Generative AI มาทำงานแทนเรา ในงานที่เราทำเป็นอยู่แล้ว แต่อยากลดเวลา และเราสามารถตรวจงานได้ง่ายครับ โดยรวมคือใช้เวลาเพียง 5-10 นาที แทนที่จะใช้ 30-60 นาที ทำให้มีเวลาไปดูพระอาทิตย์ตกทันนั่นเอง
.
โจทย์คือ API Security จุดซึ่งมีการทำ End to end encryption ตัว Value ใน Key: Value เช่นยอดเงินในบัญชีธนาคาร
.
การจะทำ Security ให้กับ API ได้ โดยปกติก็จะต้องมี API ใช้งานก่อน หรือก็คือมีข้อมูล API มาให้ปกป้องนั่นเอง ฉะนั้นเราจะสร้าง API Application ขึ้นมา 1 ตัว โดยมีความสามารถในการ Encrypted ยอดเงินในบัญชีได้ด้วย ฉะนั้้นเมื่อปลายทางอ่านข้อมูลนี้ไปได้ แบบ End to end ก็จะได้ Encrypted value ซึ่งจะต้องนำไป decrypt ด้วย key เดียวกัน จึงจะได้ยอดเงินจริงออกมาใช้งานได้นั่นเอง
.
โดยปกติแล้ว เราสามารถสร้าง Rest API ด้วย Python ได้เลย และเรียกใช้ .encrypt function เพื่อทำการ encrypt ข้อมูลที่ต้องการ แต่วันนี้เราจะลองใช้ Generative Model มาช่วยเขียน

ผลที่ได้ในแต่ละ Step อาจมีความต่างกันบ้าง เช่น Logic ที่ไม่ถูก อันนี้ต้องดูแล้วปรับเฉพาะจุด

และความต่างเรื่องการตั้งชื่อตัวแปร ว่าสอดคล้องกับความหมายมากน้อยแค่ไหน

อันนี้เป็นผลลัพธ์ที่ได้ เมื่อเรียกไปที่ /customer/1

จะพบว่า Account_number และ balance ถูก Encrypted เรียบร้อยแล้ว

สามารถปรับแต่งให้บางจุด เกิดความไม่ปลอดภัย เพื่อทดสอบการทำ API Security ได้ด้วย
เช่นที่ /customer/3 จะไม่ถูก encrypted

จะพบว่า Account_number และ balance ไม่ได้ถูก Encrypted ไว้

โดยเบื้องหลัง จะมี Function สำหรับ Encrypt
การใช้งานจริงจำเป็นต้องปรับเปลี่ยนวิธีการเก็บ Encryption key ให้ปลอดภัย

# === MOCK AES KEY (32 bytes = AES-256) ===
AES_KEY = b"this_is_a_32_byte_secret_key!!"

def encrypt_aes(data: str) -> str:
    cipher = AES.new(AES_KEY, AES.MODE_GCM)
    ciphertext, tag = cipher.encrypt_and_digest(data.encode())

    encrypted_payload = {
        "nonce": base64.b64encode(cipher.nonce).decode(),
        "ciphertext": base64.b64encode(ciphertext).decode(),
        "tag": base64.b64encode(tag).decode()
    }

    return base64.b64encode(json.dumps(encrypted_payload).encode()).decode()

โดยใน Function จะมีการเรียก .encrypt_and_digest และ .b64encode ตามลำดับ

หลังจากเพิ่ม endpoint เขาไป 5-6 ตัว จะได้ไฟล์ล่าสุด main.py

from fastapi import FastAPI
from Crypto.Cipher import AES
import base64
import json

app = FastAPI()

# === MOCK AES KEY (32 bytes = AES-256) ===
AES_KEY = b"this_is_a_32_byte_secret_key!!"

def encrypt_aes(data: str) -> str:
    cipher = AES.new(AES_KEY, AES.MODE_GCM)
    ciphertext, tag = cipher.encrypt_and_digest(data.encode())

    encrypted_payload = {
        "nonce": base64.b64encode(cipher.nonce).decode(),
        "ciphertext": base64.b64encode(ciphertext).decode(),
        "tag": base64.b64encode(tag).decode()
    }

    return base64.b64encode(json.dumps(encrypted_payload).encode()).decode()


# =========================
# Customer Endpoints
# =========================

@app.get("/customer/1")
def get_customer_1():
    return {
        "customer_id": 1,
        "full_name": "John Smith",
        "bank_name": "Mock National Bank",
        "account_type": "Savings",
        "account_number_encrypted": encrypt_aes("123-4-56789-0"),
        "balance_encrypted": encrypt_aes("125430.75"),
        "currency": "USD",
        "status": "Active",
        "encryption": "AES-256-GCM"
    }


@app.get("/customer/2")
def get_customer_2():
    return {
        "customer_id": 2,
        "full_name": "Emma Johnson",
        "bank_name": "Mock National Bank",
        "account_type": "Checking",
        "account_number_encrypted": encrypt_aes("987-6-54321-9"),
        "balance_encrypted": encrypt_aes("98750.25"),
        "currency": "USD",
        "status": "Active",
        "encryption": "AES-256-GCM"
    }


# =========================
# Branch Endpoint (NO Encryption)
# =========================

@app.get("/branch")
def get_branches():
    return {
        "bank_name": "Mock National Bank",
        "total_branches": 3,
        "branches": [
            {
                "branch_id": 1,
                "branch_name": "Bangkok Main Branch",
                "address": "123 Silom Road, Bangkok, Thailand",
                "phone": "[phone]",
                "status": "Open"
            },
            {
                "branch_id": 2,
                "branch_name": "Chiang Mai Branch",
                "address": "88 Nimmanhaemin Road, Chiang Mai, Thailand",
                "phone": "[phone]",
                "status": "Open"
            },
            {
                "branch_id": 3,
                "branch_name": "Phuket Branch",
                "address": "55 Patong Beach Road, Phuket, Thailand",
                "phone": "[phone]",
                "status": "Closed"
            }
        ]
    }


# =========================
# Finance Endpoint (ALL Encrypted)
# =========================

@app.get("/finance")
def get_finance_summary():
    return {
        "bank_name": "Mock National Bank",
        "report_type": "Branch Deposit Summary",
        "currency": "USD",
        "encryption": "AES-256-GCM",
        "branches": [
            {
                "branch_id": 1,
                "branch_name": "Bangkok Main Branch",
                "total_deposit_encrypted": encrypt_aes("12500000.50")
            },
            {
                "branch_id": 2,
                "branch_name": "Chiang Mai Branch",
                "total_deposit_encrypted": encrypt_aes("5420000.75")
            },
            {
                "branch_id": 3,
                "branch_name": "Phuket Branch",
                "total_deposit_encrypted": encrypt_aes("3189000.00")
            }
        ]
    }

จากนั้นทำการสร้าง docker compose, พร้อม Dockerfile และสร้าง tunnel published ขึ้น https

สรุปต้นแบบใช้เวลาประมาณ 5 นาที รวมปรับเพิ่ม Endpoints ไม่เกิน 10 นาที ก็ได้ Apps ไว้ทดสอบ API Security 1 ตัว เทียบกับต้องเขียนเองหมด ต้องหาชื่อ library ที่จะใช้ อย่างเร็วก็น่าจะครึ่งชั่วโมง ถือว่าประหยัดเวลาได้ประมาณ 6 เท่านั่นเอง

ทั้งนี้ เน้นว่าเป็นงานที่เราทำเป็นอยู่แล้ว และสามารถตรวจงานได้ว่าตอบโจทย์ที่เราต้องการไหม หรือก็คือรู้ว่าสิ่งที่ได้มาทำอะไรอยู่นั่นเอง