Python โหลด Secret ให้ครบก่อนเริ่มงาน พร้อมปิดค่าลับใน Log

ระบบรับแจ้งยอดชำระเงินเปิดติด แต่เพิ่งรู้ว่าลืมตั้งคีย์ตอนลูกค้าส่งรายการเข้ามา แบบนี้คนเฝ้าระบบต้องตามแก้ทั้ง config และงานค้างครับ ช่วงนี้มีข่าวแพลตฟอร์มแนะนำให้หมุนคีย์ทั้งหมด จึงเป็นจังหวะดีที่จะจัดทางโหลดค่าลับและเปลี่ยนคีย์ให้ชัดเจน log ที่ช่วยหาบั๊กก็อาจเก็บคีย์เสียเอง

WHY: รู้ว่าขาดอะไรตั้งแต่เริ่ม และไม่แจกคีย์ผ่าน log

Secret คือค่าลับ เช่น โทเคนยืนยันตัวตน ส่วน environment variable คือตัวแปรสภาพแวดล้อมที่โปรเซสอ่านได้ เราจะตรวจให้ครบก่อนรับงาน แบบ fail fast คือหยุดทันทีเมื่อเงื่อนไขจำเป็นไม่พร้อม

• แจ้งชื่อที่ขาดทั้งหมดในรอบเดียว ลดการเปิดบริการซ้ำ
• ห่อค่าลับก่อนส่งต่อ เพื่อลดการเผลอพิมพ์ค่าจริงเมื่อดู config
• เตรียมรับคีย์ปัจจุบันและคีย์ก่อนหน้า เพื่อทยอยเปลี่ยนผู้ส่ง

ทำงานยังไง

ตาม เอกสาร os os.environ เป็น mapping หรือโครงสร้างจับคู่ชื่อกับค่าที่เก็บสตริง การเข้าถึงชื่อที่ไม่มีด้วยวงเล็บเหลี่ยมเกิด KeyError ตามพฤติกรรม mapping ที่เทสต์ด้วย dict ส่วน os.getenv(name) คืน None เมื่อไม่มีชื่อ หรือคืนค่าเริ่มต้นที่กำหนดไว้ จึงอย่าใช้ค่าเริ่มต้นที่ดูเหมือนคีย์จริงเพื่อกลบ config ที่ขาด

ผมจะส่ง mapping เข้า loader เพื่อให้ตอนใช้งานจริงรับ os.environ แต่ตอนเทสต์รับ dict เปรียบเหมือนเช็กกุญแจก่อนเปิดร้าน: แจ้งว่าขาดกุญแจตู้ไหนได้ แต่ไม่ต้องถ่ายร่องกุญแจลงสมุดเวรครับ ในตัวอย่างนี้ค่าที่หาย ว่าง หรือมีแต่ช่องว่างถือว่าใช้ไม่ได้ แต่ค่าที่ผ่านจะถูกเก็บตามต้นฉบับ ไม่ตัดช่องว่างจนคีย์เปลี่ยน

คลาส Secret ซ่อนข้อความผ่าน __str__ และ __repr__ ส่วน logging.Filter ตรวจข้อความก่อน handler ซึ่งเป็นตัวส่ง log ออกปลายทาง ตัวกรองนี้ประกอบข้อความด้วย getMessage() แทนค่าลับที่รู้จัก แล้วล้าง args เพื่อไม่ให้จัดรูปแบบซ้ำ ดูรายละเอียดใน เอกสาร logging

โค้ดตัวอย่างพร้อมเทสต์

บันทึกเป็น secret_demo.py แล้วรัน python3 -I -B secret_demo.py ข้อมูลเป็นตัวอย่างสมมติ บริการจริงเรียก load_config(os.environ) ครั้งเดียวก่อนรับงาน ไฟล์นี้รันเฉพาะเทสต์

import io
import logging
import os
import secrets
import unittest


class Secret:
    def __init__(self, value):
        self._value = value

    def __str__(self):
        return "[REDACTED]"

    def __repr__(self):
        return "Secret([REDACTED])"

    def reveal(self):
        return self._value


REQUIRED = ("PAYMENT_KEY", "WEBHOOK_KEY")


def load_config(env):
    missing = [name for name in REQUIRED
               if not env.get(name, "").strip()]
    if missing:
        raise ValueError("Missing secrets: " + ", ".join(missing))
    config = {name: Secret(env[name]) for name in REQUIRED}
    previous = env.get("WEBHOOK_KEY_PREVIOUS", "")
    if previous.strip():
        config["WEBHOOK_KEY_PREVIOUS"] = Secret(previous)
    return config


def accepts_token(candidate, config):
    # Compare bytes so Thai text is supported too.
    supplied = candidate.encode("utf-8")
    matched = False
    for name in ("WEBHOOK_KEY", "WEBHOOK_KEY_PREVIOUS"):
        if name in config:
            expected = config[name].reveal().encode("utf-8")
            same = secrets.compare_digest(supplied, expected)
            matched = matched | same
    return matched


class RedactSecrets(logging.Filter):
    def __init__(self, secret_objects):
        super().__init__()
        values = {item.reveal() for item in secret_objects}
        self.values = sorted((v for v in values if v),
                             key=len, reverse=True)

    def filter(self, record):
        message = record.getMessage()
        for value in self.values:
            message = message.replace(value, "[REDACTED]")
        record.msg = message
        record.args = ()
        return True


class SecretTests(unittest.TestCase):
    def env(self):
        return {"PAYMENT_KEY": "demo-pay-20261023",
                "WEBHOOK_KEY": "demo-hook-new",
                "WEBHOOK_KEY_PREVIOUS": "demo-hook-old"}

    def test_mapping_missing_key(self):
        with self.assertRaises(KeyError):
            _ = {}["PAYMENT_KEY"]
        self.assertIsNone({}.get("PAYMENT_KEY"))

    def test_all_missing_and_blank(self):
        for env in ({}, {"PAYMENT_KEY": " ", "WEBHOOK_KEY": ""}):
            with self.subTest(env=env):
                with self.assertRaises(ValueError) as caught:
                    load_config(env)
                self.assertEqual(str(caught.exception),
                                 "Missing secrets: PAYMENT_KEY, WEBHOOK_KEY")

    def test_error_has_names_only(self):
        with self.assertRaises(ValueError) as caught:
            load_config({"PAYMENT_KEY": "demo-private-value"})
        self.assertEqual(str(caught.exception),
                         "Missing secrets: WEBHOOK_KEY")

    def test_wrapping_and_original_value(self):
        env = self.env()
        env["PAYMENT_KEY"] = " demo-pay "
        config = load_config(env)
        self.assertEqual(str(config["PAYMENT_KEY"]), "[REDACTED]")
        self.assertEqual(repr(config["PAYMENT_KEY"]),
                         "Secret([REDACTED])")
        self.assertEqual(config["PAYMENT_KEY"].reveal(), " demo-pay ")
        self.assertNotIn("demo-pay", repr(config))

    def test_rotation_and_startup_snapshot(self):
        env = self.env()
        config = load_config(env)
        self.assertTrue(accepts_token("demo-hook-new", config))
        self.assertTrue(accepts_token("demo-hook-old", config))
        self.assertFalse(accepts_token("ผิดคีย์", config))
        env.pop("WEBHOOK_KEY_PREVIOUS")
        self.assertTrue(accepts_token("demo-hook-old", config))
        fresh = load_config(env)
        self.assertFalse(accepts_token("demo-hook-old", fresh))
        env["WEBHOOK_KEY_PREVIOUS"] = " "
        self.assertNotIn("WEBHOOK_KEY_PREVIOUS", load_config(env))

    def test_redacts_message_and_arguments(self):
        env = self.env()
        config = load_config(env)
        stream = io.StringIO()
        handler = logging.StreamHandler(stream)
        handler.setFormatter(logging.Formatter("%(message)s"))
        handler.addFilter(RedactSecrets(config.values()))
        logger = logging.getLogger("showkhun.secret-demo")
        logger.setLevel(logging.INFO)
        logger.propagate = False
        logger.addHandler(handler)
        try:
            logger.info("สมชาย ยอด 1250 บาท วันที่ 2026-10-23")
            logger.info("config=%r", config)
            # Deliberate mistakes to test the safety net.
            logger.info("current=%s old=%s", env["WEBHOOK_KEY"],
                        env["WEBHOOK_KEY_PREVIOUS"])
            logger.info("accidental=" + env["PAYMENT_KEY"])
        finally:
            logger.removeHandler(handler)
            handler.close()
        output = stream.getvalue()
        for value in env.values():
            self.assertNotIn(value, output)
        self.assertIn("[REDACTED]", output)
        self.assertIn("สมชาย ยอด 1250 บาท", output)


if __name__ == "__main__": unittest.main()

ผมรันตัวอย่างนี้ด้วย Python 3.14.7 แล้ว เทสต์ทั้ง 6 รายการผ่านครับ

ฟังก์ชันรับโทเคนใช้ secrets.compare_digest เพื่อลดความเสี่ยงจาก timing attack หรือการเดาค่าผ่านเวลาประมวลผล โดยแปลงทั้งคู่เป็น bytes ก่อนเทียบ ไม่ได้หมายความว่าทั้งระบบใช้เวลาเท่ากันเสมอ

WEBHOOK_KEY คู่กับ WEBHOOK_KEY_PREVIOUS เป็นรูปแบบที่เราออกแบบเอง ไม่ใช่มาตรฐาน Python ตั้งคีย์ใหม่เป็นตัวหลักและคีย์เก่าเป็นตัวก่อนหน้า แล้วเริ่มโปรเซสใหม่ เมื่อผู้ส่งย้ายครบ ให้ถอดคีย์เก่าแล้วเริ่มโปรเซสใหม่อีกครั้ง ทีมต้องกำหนดวันถอดเอง โค้ดไม่มีตัวจับเวลาหมดอายุ

จุดที่คนพลาดบ่อย

• ตรวจแค่ว่ามีชื่ออยู่: ค่าว่างก็ยังผ่านได้ แก้ด้วยนโยบายตรวจค่าว่างก่อนสร้าง Secret และรวมชื่อที่ขาดให้ครบ อย่าแนบ dict ต้นทางในข้อความผิดพลาด
• คิดว่าห่อแล้วเข้ารหัส: Secret ยังเก็บค่าจริงในหน่วยความจำ และ reveal() ก็คืนค่าจริง ใช้เมธอดนี้เฉพาะจุดที่ต้องส่งคีย์หรือเทียบค่า ไม่เอาผลไปทำข้อความ debug
• ติด filter ผิดจุด: filter ของ logger บรรพบุรุษไม่กรอง record ที่ส่งต่อจากลูกตาม เอกสาร logging จึงติดทุก handler ก่อนส่งออกหรือเข้าคิว
• หวังให้ filter จับทุกอย่าง: ตัวอย่างแทนเฉพาะค่าตรงตัวในข้อความรวม msg กับ args ไม่ครอบคลุม traceback, ฟิลด์ extra, ค่าที่ถูกเข้ารหัส หรือแค่บางส่วนของคีย์ แก้หลัก ๆ ด้วยการไม่ใส่ค่าลับลง log ตั้งแต่ต้น แล้วตรวจเส้นทาง log เพิ่มตามรูปแบบที่ระบบใช้
• เปลี่ยนค่าภายนอกแล้วรอให้แอปเห็นเอง: เราอ่านครั้งเดียวตอนเริ่ม และ os.environ มีพฤติกรรมเก็บ mapping ตามที่ เอกสาร os อธิบาย จึงส่ง config ใหม่และเริ่มโปรเซสใหม่ตามขั้นตอน deploy พร้อมสร้าง filter ใหม่จากคีย์ที่โหลด อย่าเก็บคีย์เก่าไว้ไม่มีกำหนด

ใช้ตรงไหนในงานจริง

• บริการรับ webhook ตรวจคีย์ก่อนเปิดรับคำขอ แล้วค่อยรองรับคีย์ก่อนหน้าระหว่างย้ายผู้ส่ง
• งานสรุปยอดรายวัน หยุดก่อนดึงข้อมูลเมื่อคีย์ไม่ครบ
• เครื่องมือภายใน แจ้งชื่อ config ที่ต้องแก้โดยไม่แสดงค่าจริง
• ชุดเทสต์ ใช้ dict จำลองค่าที่ขาดและการหมุนคีย์ โดยไม่แตะ environment จริง

เทียบกับทางเลือกอื่น

approach	ดีตรงไหน	ระวังอะไร
os.environ[name] ทีละจุด	บังคับว่าต้องมีชื่อ	เจอ KeyError ทีละชื่อ และอาจเกิดหลังเริ่มรับงาน
os.getenv(name, default)	เหมาะกับค่าที่เลือกไม่ตั้งได้	ค่าเริ่มต้นอาจซ่อนการลืมตั้ง secret
Loader รวม + Secret + filter	ตรวจครบก่อนเริ่ม ลดการเผลอแสดงค่า	ต้องติด filter ทุกทางออก และกำหนดวันเลิกใช้คีย์เก่า

Recap

• รวมชื่อ secret ที่จำเป็นแล้วตรวจให้ครบก่อนรับงาน
• รายงานเฉพาะชื่อที่ขาด ไม่รายงานค่าหรือ environment ทั้งชุด
• ปฏิเสธค่าว่างตามนโยบายโดยไม่แก้เนื้อคีย์ที่รับมา
• ใช้ Secret ลดการเผลอแสดงค่า และจำกัดจุดเรียก reveal()
• ติด filter ทุก handler พร้อมรู้ขอบเขตของการแทนข้อความ
• เทสต์ด้วย dict ทั้งกรณีคีย์ขาด คีย์เก่า และ log หลุด
• กำหนดช่วงรับคีย์เก่า แล้วถอดออกพร้อมเริ่มโปรเซสใหม่