Flask ও FastAPI-তে এসকিউএল ইনজেকশন প্রতিরোধের উপায়

পাইথনের মাইক্রোফ্রেমওয়ার্ক Flask এবং আধুনিক অ্যাসিনক্রোনাস ফ্রেমওয়ার্ক FastAPI বর্তমানে মাইক্রোসার্ভিস, রিয়েল-টাইম সার্ভিস ও মেশিন লার্নিং এপিআই তৈরিতে সর্বাধিক ব্যবহৃত হয়। ফুল-স্ট্যাক ফ্রেমওয়ার্ক যেমন Django-এর মতো এগুলোতে কোনো পূর্বনির্ধারিত ডাটাবেস লেয়ার থাকে না; বরং ডেভেলপাররা সাধারণত SQLAlchemy, asyncpg কিংবা databases লাইব্রেরি দিয়ে নিজস্ব ডাটাবেস আর্কিটেকচার তৈরি করেন।

এই স্বাধীনতা থাকার কারণে ডেভেলপাররা যদি সরাসরি র এসকিউএল (Raw SQL) কুয়েরিতে স্ট্রিং কনক্যাটেনেশন ব্যবহার করেন, তবে ভয়াবহ এসকিউএল ইনজেকশন (SQLi) ঝুঁকি দেখা দেয়।

আক্রমণের মূল কার্যপ্রণালী ও মেকানিজম বুঝতে আমাদের এসকিউএল ইনজেকশন পরিচিতি ও প্রতিরোধ নির্দেশিকা এবং মনোলিথিক পাইথন অ্যাপ্লিকেশনের জন্য Django-তে এসকিউএল ইনজেকশন প্রতিরোধ গাইড পড়ে দেখতে পারেন।

এই গাইডে আধুনিক SQLAlchemy 2.0, নিরাপদ text() ফাংশন এবং Pydantic ভ্যালিডেশন ব্যবহারের মাধ্যমে কীভাবে Flask ও FastAPI অ্যাপ সুরক্ষিত রাখতে হয় তা বিস্তারিত আলোচনা করা হয়েছে।

Flask ও FastAPI-তে এসকিউএল ইনজেকশনের ঝুঁকি কীভাবে আসে

Flask-এ ক্লায়েন্টের ডেটা আসে request.args বা request.json থেকে এবং FastAPI-তে আসে পাথ, কুয়েরি প্যারামিটার বা Pydantic মডেল থেকে। যখন এই ইনপুট সরাসরি এসকিউএল স্ট্রিংয়ে বসানো হয়:

# বিপজ্জনক: Flask রাউটে পাইথন স্ট্রিং ফরম্যাটিং
@app.route('/api/users/profile')
def user_profile():
    user_id = request.args.get('id')
    
    # স্ট্রিং ইন্টারপোলেশন ব্যবহার করায় মারাত্মক ইনজেকশন সম্ভব
    query = f"SELECT id, username, email FROM users WHERE id = {user_id}"
    result = db.session.execute(text(query))
    return jsonify([dict(row) for row in result])

ব্যবহারকারী যদি 1 OR 1=1 পাঠায়, তবে ডাটাবেসের সম্পূর্ণ ইউজার তথ্য প্রকাশ পেয়ে যাবে। একই সমস্যা FastAPI অ্যাপ্লিকেশনেও হতে পারে যদি সরাসরি আন-প্যারামিটারাইজড স্ট্রিং এক্সিকিউট করা হয়।

SQLAlchemy 2.0: প্যারামিটারাইজেশনের সেরা মান

আধুনিক SQLAlchemy 2.0-তে র কুয়েরি লেখার জন্য text() ফাংশন ব্যবহার করা হয়। text() ব্যবহারের সময় স্ট্রিং ফরম্যাটিং না করে সর্বদা নেইমড প্লেসহোল্ডার (:param) ব্যবহার করতে হবে:

# নিরাপদ: FastAPI-তে SQLAlchemy 2.0 প্যারামিটারাইজড text()
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
from pydantic import BaseModel
from database import get_async_db

app = FastAPI()

@app.get('/api/users/{user_id}')
async def get_user(user_id: int, db: AsyncSession = Depends(get_async_db)):
    # নেইমড প্লেসহোল্ডার :id ব্যবহার করে প্যারামিটারাইজড কোয়েরি
    stmt = text("SELECT id, username, email FROM users WHERE id = :id AND is_active = :active")
    
    # ডিকশনারির মাধ্যমে সুরক্ষিতভাবে ভ্যালু বাইন্ড করা
    result = await db.execute(stmt, {"id": user_id, "active": True})
    user = result.mappings().first()
    
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ইউজার পাওয়া যায়নি")
        
    return user

এই পদ্ধতিতে ডাটাবেস ইঞ্জিন আগে কুয়েরি স্ট্রাকচার কম্পাইল করে এবং :id ভ্যালুকে শুধুমাত্র লিটারেল ডেটা হিসেবে মূল্যায়ন করে।

SQLAlchemy 2.0 ওআরএম Select সিনট্যাক্স

সরাসরি টেক্সট কুয়েরি না লিখে ওআরএম ব্যবহার করলে SQLAlchemy স্বয়ংক্রিয়ভাবে কুয়েরি প্যারামিটারাইজেশন পরিচালনা করে:

# নিরাপদ: SQLAlchemy 2.0 Select মেথড
from sqlalchemy import select
from models import User

async def find_user_by_email(db: AsyncSession, email: str):
    stmt = select(User).where(User.email == email, User.is_active.is_(True))
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

Pydantic দিয়ে প্রাথমিক ইনপুট ভ্যালিডেশন

FastAPI-তে Pydantic ডেটাবেসে পৌঁছানোর আগেই ইনপুট যাচাই করার প্রথম স্তরের ফিল্টার হিসেবে কাজ করে:

from pydantic import BaseModel, Field

class ProductFilterSchema(BaseModel):
    category: str = Field(..., min_length=2, max_length=50)
    min_price: float = Field(default=0.0, ge=0)
    max_price: float = Field(default=10000.0, ge=0)
    
    class Config:
        frozen = True

যদি কেউ নিউমেরিক প্রাইসের জায়গায় কোনো ক্ষতিকর স্ট্রিং পাঠানোর চেষ্টা করে, FastAPI স্বয়ংক্রিয়ভাবে HTTP 422 Unprocessable Entity এরর দিয়ে রিকোয়েস্ট আটকে দেয়।

ডায়নামিক কলাম সর্টিংয়ের নিরাপদ কৌশল

ডাটাবেস ড্রাইভার কলাম নেম বা ASC/DESC কি-ওয়ার্ড প্যারামিটার হিসেবে গ্রহণ করে না। তাই ডায়নামিক সর্টিংয়ের ক্ষেত্রে কঠোর পাইথন ডিকশনারি হোয়াইটলিস্ট ব্যবহার করতে হবে:

# নিরাপদ: কলামের নামের হোয়াইটলিস্ট ম্যাপিং
SORTABLE_COLUMNS = {
    "price": "price",
    "name": "name",
    "created": "created_at"
}

def build_secure_order_query(sort_by: str, direction: str = "desc"):
    column = SORTABLE_COLUMNS.get(sort_by, "created_at")
    dir_clause = "DESC" if direction.lower() == "desc" else "ASC"
    
    # ভ্যালিডেট করা কলাম এবং দিক স্ট্রিংয়ে বসানো নিরাপদ
    return text(f"SELECT id, name, price FROM items ORDER BY {column} {dir_clause} LIMIT 50")

প্রজেক্টের নিয়মিত কোডবেস মান উন্নত করতে আমাদের ক্লিন কোড ও কোড রিভিউ নীতিমালা দেখুন।

সচরাচর যেসব ভুল পরিলক্ষিত হয়

ভুল পদ্ধতি	নিরাপত্তা ঝুঁকি	নিরাপদ সমাধান
text() এর মধ্যে f-string ব্যবহার	র এসকিউএল সরাসরি ইনজেকশনের শিকার হয়	:named_param প্লেসহোল্ডার এবং ডিকশনারি আর্গুমেন্ট ব্যবহার করুন
সরাসরি session.execute(f"...")	ডাটাবেস এস্কেপিং সম্পূর্ণ অকার্যকর হয়ে যায়	কুয়েরি সর্বদা text() দিয়ে র‍্যাপ করে প্যারামিটার বাইন্ড করুন
কলাম বা টেবিল নেমে সরাসরি ইউজার ইনপুট	প্যারামিটার বাইন্ডিং আইডেন্টিফায়ারে কাজ করে না	পাইথন set বা dict দিয়ে কলামের নাম হোয়াইটলিস্ট করুন
ক্লায়েন্টে ডাটাবেস এরর ও ট্রেসব্যাক দেখানো	ডাটাবেস টেবিল স্কিমা ফাঁস হওয়ার ঝুঁকি	SQLAlchemyError হ্যান্ডল করে জেনেরিক HTTP 500 এরর দিন
ডাটাবেস ইউজারের অতিরিক্ত পারমিশন	টেবিল মুছে ফেলা বা ড্রপ করার ঝুঁকি	সংযোগে ন্যূনতম প্রয়োজনীয় পারমিশনযুক্ত ইউজার ব্যবহার করুন

প্রোডাকশন অডিট চেকলিস্ট

সার্ভিস প্রোডাকশনে তোলার আগে নিচের বিষয়গুলো মিলিয়ে নিন:

• সকল র এসকিউএল স্টেটমেন্ট text() দিয়ে র‍্যাপ করা এবং নেইমড প্যারামিটার (:key) ব্যবহার করা হয়েছে।
• FastAPI রাউট আর্গুমেন্ট ও রিকোয়েস্ট বডি Pydantic স্কিমা দ্বারা কঠোরভাবে ভ্যালিডেট করা।
• ডায়নামিক সর্ট কলাম ও টেবিলের নাম নির্দিষ্ট হোয়াইটলিস্ট ডিকশনারির বিপরীতে যাচাইকৃত।
• কাস্টম এক্সেপশন হ্যান্ডলার ডাটাবেসের অভ্যন্তরীণ এরর ক্লায়েন্টে যাওয়া রোধ করছে।
• সিআই পাইপলাইনে Bandit বা Semgrep-এর মতো SAST টুল দিয়ে স্ট্রিং কনক্যাটেনেশন স্ক্যান করা হচ্ছে।

সারসংক্ষেপ

Flask এবং FastAPI অ্যাপ্লিকেশনে এসকিউএল ইনজেকশন রোধ করতে আধুনিক SQLAlchemy 2.0-এর প্যারামিটারাইজড কুয়েরি মেনে চলা অপরিহার্য। Pydantic দিয়ে কড়া ইনপুট ফিল্টারিং এবং কলাম নেমে হোয়াইটলিস্ট নীতি নিশ্চিত করলে আপনার পাইথন মাইক্রোসার্ভিস সম্পূর্ণ সুরক্ষিত থাকবে।

তথ্যসূত্র: SQLAlchemy 2.0 Unified Tutorial on Working with Engines and Connections, FastAPI Security Guidelines, OWASP Python Security Cheat Sheet, CWE-89 (মূল্যায়ন নভেম্বর ২০২৬)।