SiamCafe · Blog
Stable Diffusion ComfyUI DevSecOps Integration —
บทความ

Stable Diffusion ComfyUI DevSecOps Integration —

เผยแพร่ 28 พฤษภาคม 2569

ComfyUI DevSecOps

Stable Diffusion ComfyUI Node-based Workflow DevSecOps CI/CD Security Model Management SDXL ControlNet LoRA API Automation Container Docker

ToolTypeAPINode-basedเหมาะกับ
ComfyUINode GUIREST+WSใช่Advanced/Production
Automatic1111Web GUIRESTไม่เริ่มต้น
InvokeAIWeb GUIRESTไม่Creative
FooocusSimple GUIไม่มีไม่ง่ายสุด

ComfyUI Setup และ API

=== ComfyUI Setup & API ===

Docker

docker run -d --gpus all \

-p 8188:8188 \

-v ./models:/comfyui/models \

-v ./output:/comfyui/output \

comfyui/comfyui:latest

ComfyUI API — Generate Image

import json

import requests

import websocket

import uuid

COMFYUI_URL = "http://localhost:8188"

client_id = str(uuid.uuid4())

def queue_prompt(workflow):

p = {"prompt": workflow, "client_id": client_id}

response = requests.post(f"{COMFYUI_URL}/prompt", json=p)

return response.json()

def get_image(filename, subfolder, folder_type):

params = {"filename": filename, "subfolder": subfolder, "type": folder_type}

response = requests.get(f"{COMFYUI_URL}/view", params=params)

return response.content

# Load Workflow JSON

with open("workflow_api.json") as f:

workflow = json.load(f)

# Modify Prompt

workflow["6"]["inputs"]["text"] = "a beautiful sunset over mountains"

workflow["3"]["inputs"]["seed"] = 42

# Queue and Wait

result = queue_prompt(workflow)

prompt_id = result["prompt_id"]

# WebSocket for progress

ws = websocket.WebSocket()

ws.connect(f"ws://localhost:8188/ws?clientId={client_id}")

while True:

msg = json.loads(ws.recv())

if msg["type"] == "executing" and msg["data"]["node"] is None:

break # Done

from dataclasses import dataclass

from typing import List

@dataclass

class WorkflowNode:

node_id: str

node_type: str

purpose: str

nodes = [

WorkflowNode("1", "CheckpointLoaderSimple", "Load SD Model"),

WorkflowNode("2", "CLIPTextEncode", "Positive Prompt"),

WorkflowNode("3", "CLIPTextEncode", "Negative Prompt"),

WorkflowNode("4", "KSampler", "Generate Latent"),

WorkflowNode("5", "VAEDecode", "Decode to Image"),

WorkflowNode("6", "SaveImage", "Save Output"),

WorkflowNode("7", "ControlNetApply", "ControlNet Guidance"),

WorkflowNode("8", "LoraLoader", "Load LoRA Weights"),

]

print("=== ComfyUI Workflow Nodes ===")

for n in nodes:

print(f" [Node {n.node_id}] {n.node_type} — {n.purpose}")

DevSecOps Pipeline

=== DevSecOps Pipeline for AI Image Gen ===

GitHub Actions CI/CD

.github/workflows/ai-pipeline.yml

name: AI Image Pipeline

on:

push: { branches: [main] }

jobs:

security-scan:

runs-on: ubuntu-latest

steps:

  • uses: actions/checkout@v4
  • name: Scan Docker Image

uses: aquasecurity/trivy-action@master

with:

image-ref: 'comfyui:latest'

severity: 'CRITICAL, HIGH'

  • name: Model Checksum Verify

run: |

sha256sum -c models/checksums.sha256

  • name: NSFW Filter Test

run: python tests/test_content_safety.py

deploy:

needs: security-scan

runs-on: ubuntu-latest

steps:

  • name: Deploy ComfyUI

run: |

docker-compose -f docker-compose.prod.yml up -d

  • name: Health Check

run: |

curl -f http://localhost:8188/system_stats

Model Security Scanning

import hashlib

import os

def verify_model(model_path, expected_hash):

sha256 = hashlib.sha256()

with open(model_path, 'rb') as f:

for chunk in iter(lambda: f.read(8192), b''):

sha256.update(chunk)

actual = sha256.hexdigest()

if actual != expected_hash:

raise SecurityError(f"Model hash mismatch: {actual}")

return True

security_checks = {

"Container Scan": {"tool": "Trivy", "frequency": "Every Build", "severity": "Critical+High"},

"Model Checksum": {"tool": "SHA256", "frequency": "Every Load", "severity": "Critical"},

"NSFW Filter": {"tool": "Safety Checker", "frequency": "Every Generation", "severity": "High"},

"Access Control": {"tool": "OAuth2/API Key", "frequency": "Every Request", "severity": "High"},

"Audit Log": {"tool": "ELK Stack", "frequency": "Continuous", "severity": "Medium"},

"License Check": {"tool": "Custom Script", "frequency": "Model Update", "severity": "Medium"},

}

print("\nDevSecOps Security Checks:")

for check, info in security_checks.items():

print(f" [{info['severity']}] {check}")

print(f" Tool: {info['tool']} | Frequency: {info['frequency']}")

Production Architecture

=== Production Architecture ===

docker-compose.prod.yml

version: '3.8'

services:

comfyui:

image: comfyui:latest

deploy:

resources:

reservations:

devices:

  • capabilities: [gpu]

ports:

  • "8188:8188"

volumes:

  • ./models:/comfyui/models:ro
  • ./output:/comfyui/output

environment:

  • COMFYUI_LISTEN=0.0.0.0

api-gateway:

image: nginx:alpine

ports:

  • "443:443"

volumes:

  • ./nginx.conf:/etc/nginx/nginx.conf

redis:

image: redis:7-alpine

ports:

  • "6379:6379"

monitoring:

image: grafana/grafana:latest

ports:

  • "3000:3000"

architecture = {

"API Gateway": "Nginx + Rate Limiting + Auth",

"Queue": "Redis Queue สำหรับ Job Management",

"ComfyUI Workers": "GPU Nodes รัน Stable Diffusion",

"Model Registry": "S3/HuggingFace เก็บ Models",

"Storage": "S3 เก็บ Output Images",

"CDN": "CloudFront/CloudFlare ส่งภาพ",

"Monitoring": "Grafana + Prometheus Metrics",

"Logging": "ELK Stack Audit Trail",

}

print("Production Architecture:")

for component, desc in architecture.items():

print(f" [{component}]: {desc}")

Cost Estimation

costs = {

"GPU Server (A100)": "$2.50/hr",

"Storage (S3 1TB)": "$23/mo",

"CDN (10TB transfer)": "$85/mo",

"Monitoring": "$50/mo",

"Total (24/7)": "~$1,958/mo",

}

print(f"\n\nCost Estimation:")

for item, cost in costs.items():

print(f" {item}: {cost}")

เคล็ดลับ

  • Headless: รัน ComfyUI --listen 0.0.0.0 สำหรับ API Server
  • Checksum: ตรวจ SHA256 ทุก Model ก่อนใช้งาน
  • Queue: ใช้ Redis Queue จัดการ Job ไม่ให้ GPU Overload
  • NSFW: เปิด Safety Checker ทุก Generation ใน Production
  • Cache: Cache ผลลัพธ์ที่ Prompt เหมือนกัน ประหยัด GPU

ComfyUI คืออะไร

Node-based GUI Stable Diffusion Workflow SDXL ControlNet LoRA API JSON Automation Headless Production ยืดหยุ่น