Snyk Code Security กับ Data Pipeline ETL — วิธีสแกนช่องโหว่และรักษาความปลอดภัยใน ETL Pipeline | SiamCafe Blog

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

Snyk Code Security กับ Data Pipeline ETL — วิธีสแกนช่องโหว่และรักษาความปลอดภัยใน ETL Pipeline | SiamCafe Blog

Snyk คืออะไร

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

Secure ETL Pipeline Code

CI/CD Integration กับ Snyk

Security Best Practices สำหรับ ETL Pipeline

Snyk คืออะไร

ทำไม Data Pipeline ETL ต้องสแกน Security

Snyk สแกนอะไรได้บ้าง

วิธี Integrate Snyk กับ CI/CD ทำอย่างไร

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