Snyk Code Security Data Pipeline ETL

Snyk คืออะไร

Snyk เป็น Developer Security Platform ที่ช่วยค้นหาและแก้ไขช่องโหว่ด้านความปลอดภัยตลอด Software Development Lifecycle ครอบคลุม 4 ด้านหลัก คือ Open-source Dependencies (SCA), Source Code (SAST), Container Images และ Infrastructure as Code (IaC) จุดเด่นคือออกแบบมาสำหรับ Developer ใช้งานง่าย Integrate กับ IDE, Git และ CI/CD Pipeline ได้โดยตรง

สำหรับ Data Pipeline ETL ที่จัดการข้อมูลสำคัญขององค์กร Security เป็นสิ่งที่ละเลยไม่ได้ ช่องโหว่ใน ETL Pipeline อาจนำไปสู่ Data Breach, Credential Leak, SQL Injection หรือการถูก Exploit ผ่าน Vulnerable Dependencies การใช้ Snyk สแกนทุกส่วนของ Pipeline ช่วยลดความเสี่ยงเหล่านี้

ติดตั้งและตั้งค่า Snyk

# ติดตั้ง Snyk CLI
npm install -g snyk

# หรือใช้ pip
pip install snyk

# Login
snyk auth
# จะเปิด Browser ให้ Login กับ Snyk Account

# ตรวจสอบ Version
snyk --version

# === สแกน Python ETL Project ===

# โครงสร้าง ETL Project
etl-pipeline/
├── dags/
│   ├── extract.py
│   ├── transform.py
│   └── load.py
├── connectors/
│   ├── postgres_connector.py
│   ├── s3_connector.py
│   └── kafka_connector.py
├── tests/
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── terraform/
│   ├── main.tf
│   └── variables.tf
└── .snyk                  # Snyk Policy File

# 1. สแกน Dependencies (Open Source)
snyk test --file=requirements.txt --severity-threshold=high
# ผลลัพธ์จะแสดง Vulnerable Packages พร้อมแนะนำ Version ที่แก้ไข

# 2. สแกน Source Code (SAST)
snyk code test ./dags/ ./connectors/
# ตรวจสอบ SQL Injection, Hardcoded Credentials, Path Traversal

# 3. สแกน Docker Image
snyk container test etl-pipeline:latest --severity-threshold=high
# ตรวจสอบ CVE ใน Base Image และ Packages

# 4. สแกน Infrastructure as Code
snyk iac test ./terraform/ --severity-threshold=medium
# ตรวจสอบ Misconfiguration ใน Terraform

# 5. Monitor (ส่งผลลัพธ์ไป Snyk Dashboard)
snyk monitor --file=requirements.txt
snyk container monitor etl-pipeline:latest

# === Snyk Policy File (.snyk) ===
# ใช้ Ignore ช่องโหว่ที่ยอมรับได้
# .snyk
version: v1.5.0
ignore:
  SNYK-PYTHON-URLLIB3-6271557:
    - '*':
        reason: 'Low risk in our usage context'
        expires: '2026-06-01T00:00:00.000Z'
patch: {}

Secure ETL Pipeline Code

# connectors/secure_postgres.py
# ETL Connector ที่ปลอดภัยสำหรับ PostgreSQL
import os
import logging
from contextlib import contextmanager
from typing import Generator
import psycopg2
from psycopg2 import sql
from psycopg2.extras import RealDictCursor

logger = logging.getLogger(__name__)

class SecurePostgresConnector:
    """PostgreSQL Connector ที่ปลอดภัยสำหรับ ETL"""

    def __init__(self):
        # ดึง Credentials จาก Environment Variables (ไม่ Hardcode)
        self.config = {
            "host": os.environ["DB_HOST"],
            "port": int(os.environ.get("DB_PORT", 5432)),
            "dbname": os.environ["DB_NAME"],
            "user": os.environ["DB_USER"],
            "password": os.environ["DB_PASSWORD"],
            "sslmode": os.environ.get("DB_SSLMODE", "require"),
            "sslrootcert": os.environ.get("DB_SSL_CERT", ""),
            "connect_timeout": 10,
        }

    @contextmanager
    def get_connection(self) -> Generator:
        """Connection Pool กับ Auto-close"""
        conn = None
        try:
            conn = psycopg2.connect(**self.config)
            conn.set_session(autocommit=False)
            yield conn
            conn.commit()
        except Exception as e:
            if conn:
                conn.rollback()
            logger.error(f"Database error: {e}")
            raise
        finally:
            if conn:
                conn.close()

    def extract_data(self, table_name: str, columns: list[str],
                     where_clause: str = "", params: tuple = ()) -> list[dict]:
        """
        Extract ข้อมูลแบบ Parameterized Query (ป้องกัน SQL Injection)
        ใช้ sql.Identifier สำหรับ Table/Column Names
        ใช้ %s Parameters สำหรับ Values
        """
        # สร้าง Query แบบปลอดภัย
        col_ids = sql.SQL(", ").join([sql.Identifier(c) for c in columns])
        query = sql.SQL("SELECT {cols} FROM {table}").format(
            cols=col_ids,
            table=sql.Identifier(table_name),
        )

        if where_clause:
            # where_clause ต้องใช้ %s เป็น Placeholder เท่านั้น
            query = sql.SQL("{base} WHERE {where}").format(
                base=query,
                where=sql.SQL(where_clause),
            )

        with self.get_connection() as conn:
            with conn.cursor(cursor_factory=RealDictCursor) as cur:
                cur.execute(query, params)
                results = cur.fetchall()
                logger.info(f"Extracted {len(results)} rows from {table_name}")
                return results

    def load_data(self, table_name: str, data: list[dict],
                  on_conflict: str = "DO NOTHING") -> int:
        """
        Load ข้อมูลแบบ Batch Insert กับ Parameterized Query
        """
        if not data:
            return 0

        columns = list(data[0].keys())
        col_ids = sql.SQL(", ").join([sql.Identifier(c) for c in columns])
        placeholders = sql.SQL(", ").join([sql.Placeholder()] * len(columns))

        query = sql.SQL(
            "INSERT INTO {table} ({cols}) VALUES ({vals}) ON CONFLICT {conflict}"
        ).format(
            table=sql.Identifier(table_name),
            cols=col_ids,
            vals=placeholders,
            conflict=sql.SQL(on_conflict),
        )

        rows_inserted = 0
        with self.get_connection() as conn:
            with conn.cursor() as cur:
                for row in data:
                    values = [row[c] for c in columns]
                    cur.execute(query, values)
                    rows_inserted += cur.rowcount

        logger.info(f"Loaded {rows_inserted} rows into {table_name}")
        return rows_inserted

# connectors/secure_s3.py
# S3 Connector ที่ปลอดภัย
import os
import boto3
from botocore.config import Config

class SecureS3Connector:
    """S3 Connector กับ Encryption และ Access Control"""

    def __init__(self):
        self.client = boto3.client(
            "s3",
            region_name=os.environ.get("AWS_REGION", "ap-southeast-1"),
            config=Config(
                signature_version="s3v4",
                retries={"max_attempts": 3, "mode": "adaptive"},
            ),
        )
        self.bucket = os.environ["S3_BUCKET"]
        self.kms_key = os.environ.get("S3_KMS_KEY_ID", "")

    def upload_encrypted(self, key: str, data: bytes) -> dict:
        """Upload ด้วย Server-side Encryption (KMS)"""
        extra_args = {
            "ServerSideEncryption": "aws:kms",
        }
        if self.kms_key:
            extra_args["SSEKMSKeyId"] = self.kms_key

        return self.client.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=data,
            **extra_args,
        )

    def download(self, key: str) -> bytes:
        """Download (Auto-decrypt ถ้า Encrypted)"""
        response = self.client.get_object(Bucket=self.bucket, Key=key)
        return response["Body"].read()

CI/CD Integration กับ Snyk

# .github/workflows/security-scan.yml
# GitHub Actions — Snyk Security Scan สำหรับ ETL Pipeline

name: Security Scan
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
  schedule:
    - cron: '0 6 * * 1'  # ทุกวันจันทร์ 06:00 UTC

jobs:
  snyk-dependency-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install Dependencies
        run: pip install -r requirements.txt

      - name: Snyk Open Source Scan
        uses: snyk/actions/python@master
        env:
          SNYK_TOKEN: }
        with:
          args: --severity-threshold=high --file=requirements.txt

  snyk-code-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Snyk Code (SAST) Scan
        uses: snyk/actions/python@master
        env:
          SNYK_TOKEN: }
        with:
          command: code test
          args: --severity-threshold=high

  snyk-container-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Build Docker Image
        run: docker build -t etl-pipeline:} .

      - name: Snyk Container Scan
        uses: snyk/actions/docker@master
        env:
          SNYK_TOKEN: }
        with:
          image: etl-pipeline:}
          args: --severity-threshold=high

  snyk-iac-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Snyk IaC Scan
        uses: snyk/actions/iac@master
        env:
          SNYK_TOKEN: }
        with:
          args: --severity-threshold=medium
          file: terraform/

---
# Dockerfile ที่ปลอดภัยสำหรับ ETL
FROM python:3.12-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

FROM python:3.12-slim
# ไม่ใช้ Root User
RUN groupadd -r etl && useradd -r -g etl etl
WORKDIR /app

COPY --from=builder /root/.local /home/etl/.local
COPY . .

# ไม่เก็บ Secrets ใน Image
# ใช้ Environment Variables หรือ Secrets Manager แทน

RUN chown -R etl:etl /app
USER etl

ENV PATH="/home/etl/.local/bin:$PATH"
HEALTHCHECK --interval=30s CMD python -c "print('ok')"

CMD ["python", "-m", "dags.main"]

Security Best Practices สำหรับ ETL Pipeline

• ใช้ Parameterized Queries: ป้องกัน SQL Injection ใช้ Placeholder (%s) แทนการ Concatenate String ใน SQL Query
• ไม่ Hardcode Credentials: ใช้ Environment Variables, AWS Secrets Manager, HashiCorp Vault หรือ Kubernetes Secrets
• Encrypt Data at Rest: ใช้ Server-side Encryption (SSE-KMS) สำหรับ S3, TDE สำหรับ Database
• Encrypt Data in Transit: ใช้ SSL/TLS สำหรับ Database Connection, HTTPS สำหรับ API Calls
• Least Privilege: ให้สิทธิ์เฉพาะที่จำเป็น ETL User ไม่ควรมี DROP/ALTER Permissions
• Scan Dependencies ทุก Build: Integrate Snyk ใน CI/CD Pipeline Fail Build เมื่อพบ High/Critical CVE
• ไม่ใช้ Root ใน Container: สร้าง Non-root User ใน Dockerfile ป้องกัน Container Escape
• Audit Logging: บันทึกทุก Data Access ใครเข้าถึงข้อมูลอะไร เมื่อไร สำหรับ Compliance

Snyk คืออะไร

Snyk เป็น Developer Security Platform ค้นหาและแก้ไขช่องโหว่ใน Dependencies, Source Code, Container Images และ Infrastructure as Code ออกแบบสำหรับ Developer Integrate กับ IDE, Git และ CI/CD ได้ มี Free Plan สำหรับ Open-source Projects

สรุปและแนวทางปฏิบัติ

การรักษาความปลอดภัยใน Data Pipeline ETL ต้องครอบคลุมทุกชั้น ตั้งแต่ Dependency Scanning, Source Code Analysis, Container Security ไปจนถึง Infrastructure as Code Snyk ช่วยทำให้ทุกอย่างเป็นอัตโนมัติผ่าน CI/CD Pipeline สิ่งสำคัญคือใช้ Parameterized Queries ป้องกัน SQL Injection ไม่ Hardcode Credentials Encrypt ข้อมูลทั้ง At Rest และ In Transit ใช้ Least Privilege และ Scan ทุก Build