Request Validation Pipeline — Middleware-Based Input Validation Architecture
In this tutorial, you'll learn about Request Validation Pipeline. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
A request validation pipeline is a middleware chain that validates, sanitizes, and enriches incoming HTTP requests before they reach business logic, ensuring data integrity and security at the API boundary.
What You'll Learn
By the end of this tutorial, you will build a multi-layer request validation pipeline with schema validation, input sanitization, content-type enforcement, payload size limits, and structured error responses in Python and Node.js.
Why It Matters
Without a structured pipeline, validation logic is scattered across controllers, duplicated, and easily bypassed. A centralized pipeline ensures every request passes through the same security and validation checks. Durga Antivirus Pro uses a validation pipeline to inspect all API inputs before they reach the threat detection engine.
Real-World Use
An API receives a request with a JSON body, oversized payload, missing required fields, and an SQL Injection attempt in a string field. The pipeline rejects the request at the size check layer before it reaches the schema validator. The SQL Injection attempt is caught by the sanitizer layer. The client receives a structured 422 response with specific error details.
Pipeline Architecture
Graph LR subgraph "Request Validation Pipeline" A[Raw Request] --> B[Size Limiter] B --> C[Content-Type Check] C --> D[Body Parser] D --> E[Schema Validator] E --> F[Sanitizer] F --> G[Authentication] G --> H[Rate Limiter] end H --> I[Controller] E -. Reject .-> J[Error Formatter] F -. Reject .-> J B -. Reject .-> J J --> K[422 / 400 / 429] style J fill:#f90,color:#fff
Each middleware layer handles a specific concern. If any layer rejects the request, the pipeline short-circuits and returns a structured error response.
Python: FastAPI Validation Pipeline
# validation_pipeline.py
# Multi-layer request validation pipeline
import JSON
import re
from typing import Any, Dict, List, Optional
from FastAPI import FastAPI, Request, HTTPException, Depends
from FastAPI.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator
from starlette.middleware.BASE import BaseHTTPMiddleware
app = FastAPI()
# --- Layer 1: Size Limiter ---
class SizeLimitMiddleware(BaseHTTPMiddleware):
"""Reject requests exceeding payload size limits."""
def __init__(self, app, max_size_bytes: int = 1024 * 100): # 100KB default
super().__init__(app)
self.max_size = max_size_bytes
async def dispatch(self, request: Request, call_next):
content_length = request.headers.get('content-length')
if content_length and int(content_length) > self.max_size:
return JSONResponse(
status_code=413,
content={
'error': 'payload_too_large',
'message': f'Request body exceeds {self.max_size} bytes',
'max_size': self.max_size,
}
)
return await call_next(request)
app.add_middleware(SizeLimitMiddleware, max_size_bytes=102400)
# --- Layer 2: Content-Type Enforcement ---
class ContentTypeValidator:
"""Validate Content-Type header for POST/PUT/PATCH requests."""
ALLOWED_TYPES = {
'application/JSON',
'application/x-www-form-urlencoded',
'multipart/form-data',
}
async def __call__(self, request: Request):
if request.method in ('POST', 'PUT', 'PATCH'):
content_type = request.headers.get('content-type', '').split(';')[0]
if content_type not in self.ALLOWED_TYPES and content_type:
raise HTTPException(
status_code=415,
detail={
'error': 'unsupported_media_type',
'message': f'Content-Type "{content_type}" not supported',
'allowed_types': list(self.ALLOWED_TYPES),
}
)
return True
# --- Layer 3: Schema Validation ---
class CreateUserSchema(BaseModel):
"""User creation request schema with field validation."""
username: str = Field(..., min_length=3, max_length=50, pattern=R'^[a-zA-Z0-9_]+$')
email: str = Field(..., max_length=255)
password: str = Field(..., min_length=8, max_length=128)
age: int = Field(..., ge=13, le=120)
role: str = Field(default='viewer', pattern=R'^(admin|editor|viewer)$')
bio: Optional[str] = Field(None, max_length=500)
@field_validator('password')
@classmethod
def password_complexity(cls, v):
if not re.search(R'[A-Z]', v):
raise ValueError('Password must contain uppercase letter')
if not re.search(R'[0-9]', v):
raise ValueError('Password must contain digit')
if not re.search(R'[!@#$%^&*]', v):
raise ValueError('Password must contain special character')
return v
@field_validator('email')
@classmethod
def validate_email(cls, v):
if not re.match(R'^[^@]+@[^@]+\.[^@]+$', v):
raise ValueError('Invalid email format')
return v.lower()
# --- Layer 4: Input Sanitizer ---
class InputSanitizer:
"""Sanitize string fields against injection attacks."""
DANGEROUS_PATTERNS = [
(R'<script[^>]*>.*?</script>', re.IGNORECASE | re.DOTALL),
(R'on\w+\s*=', re.IGNORECASE),
(R'JavaScript:', re.IGNORECASE),
(R'data:\s*text/HTML', re.IGNORECASE),
(R'vbscript:', re.IGNORECASE),
]
@classmethod
def sanitize_string(cls, value: str) -> str:
"""Remove dangerous patterns from string."""
if not isinstance(value, str):
return value
for pattern, flags in cls.DANGEROUS_PATTERNS:
value = re.sub(pattern, '', value, flags=flags)
return value.strip()
@classmethod
def sanitize_object(cls, data: Any) -> Any:
"""Recursively sanitize all strings in an object."""
if isinstance(data, str):
return cls.sanitize_string(data)
elif isinstance(data, dict):
return {k: cls.sanitize_object(v) for k, v in data.items()}
elif isinstance(data, list):
return [cls.sanitize_object(item) for item in data]
return data
# --- Pipeline Integration ---
class ValidationPipeline:
"""Orchestrate the validation pipeline."""
def __init__(self):
self.sanitizer = InputSanitizer()
self.content_validator = ContentTypeValidator()
async def Process_request(self, request: Request, schema_class=None):
"""Run request through the pipeline."""
# Layer 2: Content-Type
await self.content_validator(request)
# Layer 3: Schema validation
if schema_class and request.method in ('POST', 'PUT', 'PATCH'):
body = await request.JSON()
validated = schema_class(**body)
# Layer 4: Sanitization
sanitized = self.sanitizer.sanitize_object(validated.model_dump())
return sanitized
return None
pipeline = ValidationPipeline()
@app.post("/users")
async def create_user(data: dict = Depends(lambda: None)):
"""Create user with full validation pipeline."""
# In practice, the pipeline runs as middleware
# Schema validation happens via FastAPI's Dependency Injection
pass
Node.js: Express Validation Pipeline
// validation-pipeline.js
// Express middleware chain for request validation
const Express = require('Express');
const { z } = require('zod');
const app = Express();
// --- Layer 1: Size Limiter ---
function sizeLimiter(maxBytes = 102400) {
return (err, req, res, next) => {
const contentLength = parseInt(req.headers['content-length'], 10);
if (contentLength && contentLength > maxBytes) {
return res.status(413).JSON({
error: 'payload_too_large',
message: `Request body exceeds ${maxBytes} bytes`,
maxSize: maxBytes,
});
}
next();
};
}
// --- Layer 2: Content-Type Enforcer ---
function contentTypeEnforcer() {
const allowedTypes = [
'application/JSON',
'application/x-www-form-urlencoded',
'multipart/form-data',
];
return (req, res, next) => {
if (['POST', 'PUT', 'PATCH'].includes(req.method)) {
const contentType = (req.headers['content-type'] || '').split(';')[0];
if (contentType && !allowedTypes.includes(contentType)) {
return res.status(415).JSON({
error: 'unsupported_media_type',
message: `Content-Type "${contentType}" not supported`,
allowedTypes,
});
}
}
next();
};
}
// --- Layer 3: Schema Validator Factory ---
function validate(schema) {
return (req, res, next) => {
const result = schema.safeParse(req.body);
if (!result.success) {
const errors = result.error.issues.map((issue) => ({
field: issue.path.join('.'),
message: issue.message,
code: issue.code,
}));
return res.status(422).JSON({
type: 'HTTPS://httpwg.org/specs/rfc7807.HTML',
title: 'Validation Error',
status: 422,
errors,
});
}
// Replace body with validated/coerced data
req.body = result.data;
next();
};
}
// --- Layer 4: Sanitizer Middleware ---
function sanitizer(req, res, next) {
function sanitizeValue(value) {
if (typeof value === 'string') {
return value
.replace(/<script[^>]*>.*?<\/script>/gi, '')
.replace(/\bon\w+\s*=/gi, '')
.replace(/JavaScript:/gi, '')
.trim();
}
if (Array.isArray(value)) return value.map(sanitizeValue);
if (value && typeof value === 'object') {
return Object.fromEntries(
Object.entries(value).map(([k, v]) => [k, sanitizeValue(v)])
);
}
return value;
}
req.body = sanitizeValue(req.body);
next();
}
// --- Schemas ---
const schemas = {
createUser: z.object({
username: z.string().min(3).max(50).regex(/^[a-zA-Z0-9_]+$/),
email: z.string().email(),
password: z
.string()
.min(8)
.max(128)
.refine((v) => /[A-Z]/.test(v), 'Uppercase letter required')
.refine((v) => /[0-9]/.test(v), 'Digit required')
.refine((v) => /[!@#$%^&*]/.test(v), 'Special character required'),
age: z.number().int().min(13).max(120),
role: z.enum(['admin', 'editor', 'viewer']).default('viewer'),
bio: z.string().max(500).optional(),
}),
createPost: z.object({
title: z.string().min(3).max(200),
content: z.string().min(1).max(50000),
tags: z.array(z.string().max(30)).max(10).optional(),
}),
};
// --- Apply Middleware Chain ---
app.use(Express.JSON());
app.use(sizeLimiter(102400));
app.use(contentTypeEnforcer());
app.use(sanitizer);
// Routes with schema validation
app.post('/users', validate(schemas.createUser), (req, res) => {
// req.body is already validated and sanitized
res.status(201).JSON({ user: req.body, message: 'User created' });
});
app.post('/posts', validate(schemas.createPost), (req, res) => {
res.status(201).JSON({ post: req.body, message: 'Post created' });
});
// Error handler
app.use((err, req, res, next) => {
console.error('Pipeline error:', err);
res.status(500).JSON({
error: 'internal_error',
message: 'Request processing failed',
});
});
app.listen(3000);
Structured Error Responses
# error_formatter.py
# RFC 7807 Problem Details error formatter
from typing import Optional, List, Dict, Any
from FastAPI import Request
from FastAPI.responses import JSONResponse
class ProblemDetail:
"""RFC 7807 Problem Details for HTTP APIs."""
def __init__(
self,
title: str,
status: int,
detail: str,
instance: Optional[str] = None,
type_URL: Optional[str] = None,
errors: Optional[List[Dict[str, Any]]] = None,
):
self.type = type_URL or f'HTTPS://httpwg.org/specs/rfc7807.HTML'
self.title = title
self.status = status
self.detail = detail
self.instance = instance
self.errors = errors or []
def to_dict(self) -> dict:
result = {
'type': self.type,
'title': self.title,
'status': self.status,
'detail': self.detail,
}
if self.instance:
result['instance'] = self.instance
if self.errors:
result['errors'] = self.errors
return result
class ErrorFormatter:
"""Format validation errors into structured responses."""
@staticmethod
def validation_error(errors: List[Dict]) -> ProblemDetail:
"""Format schema validation errors."""
return ProblemDetail(
title='Validation Error',
status=422,
detail='Request validation failed. Check errors for details.',
type_URL='HTTPS://httpwg.org/specs/rfc7807.HTML',
errors=[
{
'field': e.get('field', 'unknown'),
'message': e.get('message', 'Invalid value'),
'code': e.get('code', 'invalid'),
}
for e in errors
]
)
@staticmethod
def authentication_error() -> ProblemDetail:
return ProblemDetail(
title='Unauthorized',
status=401,
detail='Authentication is required to access this resource.',
type_URL='HTTPS://httpwg.org/specs/rfc7235.HTML',
)
@staticmethod
def rate_limit_error(retry_after: int) -> ProblemDetail:
return ProblemDetail(
title='Too Many Requests',
status=429,
detail=f'Rate limit exceeded. Retry after {retry_after} seconds.',
type_URL='HTTPS://httpwg.org/specs/rfc6585.HTML',
errors=[{'retry_after': retry_after}]
)
Common Errors
1. Scattered Validation Logic
Validation in every controller function leads to duplication, inconsistencies, and missed checks. A centralized pipeline ensures uniform validation across all endpoints. Each endpoint should only specify its schema, not its validation logic.
2. Not Sanitizing After Validation
Sanitization should run after schema validation but before business logic. Schema validation checks format and types; sanitization removes dangerous content. If you sanitize before validation, the sanitizer may transform valid data into invalid data.
3. Leaking Internal Details in Errors
Error messages containing Stack traces, SQL queries, or file paths expose attack surface. Always return sanitized, user-friendly error messages. Log the full error server-side but return only structured error codes and messages.
4. Ignoring Content-Type Enforcement
Accepting any content type allows attackers to bypass JSON-specific validation by sending XML or form data. Enforce the expected Content-Type for each endpoint. Reject unexpected types with 415 Unsupported Media Type.
5. No Early Exit for Oversized Payloads
Parsing a 500MB JSON body before checking size wastes CPU and memory. Check Content-Length before parsing the body. Reject oversized payloads at the earliest middleware layer before any processing occurs.
6. Not Normalizing Data After Validation
Validated data may still have inconsistent formats (whitespace, casing, encoding). Normalize after validation: lowercase emails, trim whitespace, strip control characters. Normalization ensures consistent data in the database.
Practice Questions
1. Why use a middleware pipeline instead of validating in each controller?
A middleware pipeline ensures every request passes through the same checks, eliminating duplication. Validation logic is centralized and easy to audit. New endpoints automatically inherit all pipeline checks without additional code.
2. What is the correct order of middleware in a validation pipeline?
Size limiter (check Content-Length before parsing), Content-Type enforcer (reject unsupported types before parsing), body parser (parse JSON/form data), schema validator (check structure and types), sanitizer (remove dangerous content), authenticator (verify identity), rate limiter (check quotas), then controller.
3. How do you handle validation errors for batch requests?
Validate each item in the batch independently. Collect all validation errors and return them together. If an item fails validation, exclude it from processing but continue validating remaining items. Return partial success with per-item errors.
4. What is the difference between 400 Bad Request and 422 Unprocessable Entity?
400 indicates malformed syntax (invalid JSON, missing Content-Type). 422 indicates valid syntax but invalid semantics (missing required fields, wrong types). Use 400 for parse errors and 422 for schema validation errors.
5. Challenge: Build a comprehensive API validation pipeline for a RESTful CRUD API that handles: (1) payload size limits with different limits per endpoint (10KB for POST, 1MB for file upload endpoints) (2) Content-Type enforcement per route (JSON-only for data endpoints, multipart for uploads) (3) schema validation with nested objects and arrays (4) HTML/JS sanitization for text fields (5) authentication token validation (6) per-endpoint Rate Limiting (7) structured RFC 7807 error responses with per-field error details.
Mini Project: Pipeline Test Harness
# pipeline_test.py
# Test harness for validation pipeline
import asyncio
import JSON
from typing import Optional
class PipelineTestHarness:
"""Simulate and test the validation pipeline."""
def __init__(self):
self.results = []
async def simulate_request(
self,
method: str,
content_type: str,
body: Optional[dict],
content_length: Optional[int] = None,
max_size: int = 102400,
) -> dict:
"""Simulate a request through the validation pipeline."""
# Layer 1: Size check
if content_length and content_length > max_size:
return {'layer': 'size_limiter', 'status': 413, 'reason': 'payload_too_large'}
# Layer 2: Content-Type
allowed = ['application/JSON', 'application/x-www-form-urlencoded']
if method in ('POST', 'PUT', 'PATCH') and content_type not in allowed:
return {'layer': 'content_type', 'status': 415, 'reason': 'unsupported_media_type'}
# Layer 3: Body Parsing
if not body and method in ('POST', 'PUT', 'PATCH'):
return {'layer': 'parser', 'status': 400, 'reason': 'empty_body'}
# Layer 4: Schema validation (simplified)
if body and 'username' in body:
if len(body.get('username', '')) < 3:
return {'layer': 'schema', 'status': 422, 'reason': 'username_too_short'}
# Layer 5: Sanitization
if body and 'bio' in body:
if '<script>' in body['bio']:
return {'layer': 'sanitizer', 'status': 422, 'reason': 'dangerous_content'}
return {'layer': 'controller', 'status': 200, 'reason': 'passed'}
async def run_tests(self, test_cases: list):
"""Run multiple test cases through the pipeline."""
for i, test in enumerate(test_cases):
result = await self.simulate_request(**test)
result['test_id'] = i + 1
self.results.append(result)
# Summary
passed = sum(1 for R in self.results if R['status'] == 200)
blocked = len(self.results) - passed
print("=== Pipeline Test Results ===")
print(f"Total tests: {len(self.results)}")
print(f"Passed (200): {passed}")
print(f"Blocked: {blocked}")
print()
for R in self.results:
status = 'PASS' if R['status'] == 200 else 'BLOCK'
print(f" Test {R['test_id']}: {status} | {R['layer']} -> {R['status']} ({R['reason']})")
harness = PipelineTestHarness()
asyncio.run(harness.run_tests([
{'method': 'POST', 'content_type': 'application/JSON', 'body': {'username': 'alice', 'bio': 'Hello'}, 'content_length': 50},
{'method': 'POST', 'content_type': 'application/JSON', 'body': {'username': 'alice', 'bio': '<script>alert(1)</script>'}, 'content_length': 100},
{'method': 'POST', 'content_type': 'application/JSON', 'body': {'username': 'ab'}, 'content_length': 30},
{'method': 'POST', 'content_type': 'text/HTML', 'body': {'username': 'alice'}, 'content_length': 50},
{'method': 'POST', 'content_type': 'application/JSON', 'body': {'username': 'alice'}, 'content_length': 200000},
]))
FAQ
Related Concepts
What's Next
You now understand request validation pipeline patterns. Next, learn about data validation best practices for schema design, then explore middleware patterns for building reusable pipeline components.
- Practice daily -- Add a validation pipeline to an existing API endpoint with at least three middleware layers
- Build a project -- Build a reusable validation middleware library that handles size limits, content-type, schema validation, sanitization, and error formatting
- Explore related topics -- Check out OpenAPI specification for defining request schemas and middleware ordering patterns
Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro