API Security
API Rules That Don't Leak
Web risk is rarely mysterious. It usually lies in predictable mistakes that persist under time pressure.
With API security, security only truly works when authorization is enforced explicitly per object and action.
That makes security less of a separate afterthought and more a standard quality of your product.
Immediate measures (15 minutes)
Why this matters
The core of API security is risk reduction in practice. Technical context supports the choice of measures, but implementation and enforcement are central.
Authentication
API keys vs OAuth 2.0 vs JWT
| Method | Suitable for | Risks | Recommendation |
|---|---|---|---|
| API key (header/query) | Server-to-server, internal services | No expiry, often hardcoded, difficult to rotate | Only for machine-to-machine with IP restrictions |
| OAuth 2.0 Bearer token | User-facing APIs, third-party integrations | Token theft via XSS/logs if not stored correctly | Recommended for user authentication |
| JWT (signed) | Stateless authentication, microservices | No server-side revocation, oversized tokens, algorithm confusion | Use with short exp, refresh tokens, and
alg validation |
| mTLS (client certificates) | Zero-trust, service mesh | Complex certificate management | Strongest option for service-to-service |
Implementing JWT securely
# Python — PyJWT
import jwt
from datetime import datetime, timedelta, timezone
SECRET_KEY = os.environ["JWT_SECRET"] # At least 256 bits
ALGORITHM = "HS256"
def create_token(user_id: int, role: str) -> str:
payload = {
"sub": str(user_id),
"role": role,
"iat": datetime.now(timezone.utc),
"exp": datetime.now(timezone.utc) + timedelta(minutes=15),
"iss": "api.example.com",
}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def verify_token(token: str) -> dict:
try:
# Force algorithm — prevent algorithm confusion
return jwt.decode(
token,
SECRET_KEY,
algorithms=[ALGORITHM], # NEVER algorithms=None
issuer="api.example.com",
options={"require": ["exp", "sub", "iss"]},
)
except jwt.ExpiredSignatureError:
raise AuthError("Token expired")
except jwt.InvalidTokenError:
raise AuthError("Invalid token")// Node.js — jsonwebtoken
const jwt = require('jsonwebtoken');
const SECRET = process.env.JWT_SECRET;
function verifyToken(req, res, next) {
const header = req.headers.authorization;
if (!header || !header.startsWith('Bearer ')) {
return res.status(401).json({ error: 'Token missing' });
}
try {
// Force algorithm — NEVER omit algorithms
const decoded = jwt.verify(header.slice(7), SECRET, {
algorithms: ['HS256'],
issuer: 'api.example.com',
});
req.user = decoded;
next();
} catch (err) {
return res.status(401).json({ error: 'Invalid token' });
}
}Warning: Never use
noneas an allowed algorithm. The notoriousalg: noneattack bypasses the signature entirely. Always force an explicit list of permitted algorithms.
Authorization
Authentication answers "who are you?" Authorization answers "what are you allowed to do?" The three levels of API authorization:
Object-level (BOLA/IDOR)
Broken Object Level Authorization (BOLA), also known as IDOR, is the #1 API vulnerability in the OWASP API Security Top 10.
# WRONG — no authorization check at object level
@app.get('/api/orders/<int:order_id>')
def get_order(order_id):
order = Order.query.get_or_404(order_id)
return jsonify(order.to_dict()) # Any user can request any order
# CORRECT — check ownership
@app.get('/api/orders/<int:order_id>')
@login_required
def get_order(order_id):
order = Order.query.get_or_404(order_id)
if order.user_id != current_user.id:
abort(403)
return jsonify(order.to_dict())Function-level
# Decorator for role-based access
from functools import wraps
def require_role(*roles):
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
if current_user.role not in roles:
abort(403)
return f(*args, **kwargs)
return wrapper
return decorator
@app.delete('/api/users/<int:user_id>')
@login_required
@require_role('admin')
def delete_user(user_id):
# Only admins can delete users
...Field-level
# Restrict which fields are visible per role
def serialize_user(user, viewer_role):
base = {
"id": user.id,
"name": user.name,
"email": user.email,
}
if viewer_role == "admin":
base["role"] = user.role
base["last_login_ip"] = user.last_login_ip
base["created_at"] = user.created_at.isoformat()
return baseRule: Check authorization at every level. A user who is allowed to call an endpoint does not automatically have access to every object within that endpoint.
Rate limiting & throttling
Without rate limiting your API is vulnerable to brute-force attacks, credential stuffing, and denial-of-service.
Flask-Limiter
from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
app = Flask(__name__)
limiter = Limiter(
app=app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"],
storage_uri="redis://localhost:6379", # Use Redis in production
)
# Global limit applies to all routes
# Stricter limit on authentication endpoints
@app.post('/api/auth/login')
@limiter.limit("5 per minute")
def login():
...
# Higher limit for authenticated read endpoints
@app.get('/api/products')
@limiter.limit("100 per minute")
def list_products():
...
# No limit for health checks
@app.get('/health')
@limiter.exempt
def health():
return {"status": "ok"}express-rate-limit (Node.js)
const rateLimit = require('express-rate-limit');
const RedisStore = require('rate-limit-redis');
// Global limit
const globalLimiter = rateLimit({
windowMs: 15 * 60 * 1000, // 15 minutes
max: 100, // Max 100 requests per window
standardHeaders: true, // RateLimit-* headers
legacyHeaders: false,
message: { error: 'Too many requests, please try again later' },
store: new RedisStore({ /* Redis config */ }),
});
// Strict limit on login
const loginLimiter = rateLimit({
windowMs: 15 * 60 * 1000,
max: 5,
message: { error: 'Too many login attempts' },
});
app.use('/api/', globalLimiter);
app.post('/api/auth/login', loginLimiter, authController.login);Rate limit headers
Always include rate limit information in the response:
| Header | Purpose |
|---|---|
RateLimit-Limit |
Maximum number of requests in window |
RateLimit-Remaining |
Remaining requests in current window |
RateLimit-Reset |
Unix timestamp when the window resets |
Retry-After |
Seconds until the client may try again (on 429) |
Input validation at API level
Schema validation
Validate every request against a schema. Never accept unvalidated JSON.
Pydantic (Python/FastAPI/Flask)
from pydantic import BaseModel, Field, field_validator
from typing import Optional
from enum import Enum
class Priority(str, Enum):
low = "low"
medium = "medium"
high = "high"
critical = "critical"
class CreateFindingRequest(BaseModel):
title: str = Field(min_length=3, max_length=200)
description: str = Field(max_length=10000)
severity: Priority
cvss_score: Optional[float] = Field(default=None, ge=0.0, le=10.0)
affected_hosts: list[str] = Field(max_length=100)
@field_validator('affected_hosts')
@classmethod
def validate_hosts(cls, v):
import re
for host in v:
if not re.fullmatch(r'[\w.\-:]+', host):
raise ValueError(f"Invalid hostname: {host}")
return vJoi (Node.js/Express)
const Joi = require('joi');
const createFindingSchema = Joi.object({
title: Joi.string().min(3).max(200).required(),
description: Joi.string().max(10000).required(),
severity: Joi.string().valid('low', 'medium', 'high', 'critical').required(),
cvss_score: Joi.number().min(0).max(10).precision(1).optional(),
affected_hosts: Joi.array()
.items(Joi.string().pattern(/^[\w.\-:]+$/))
.max(100)
.required(),
});
// Middleware for validation
function validate(schema) {
return (req, res, next) => {
const { error, value } = schema.validate(req.body, {
abortEarly: false,
stripUnknown: true, // Remove unknown fields
});
if (error) {
return res.status(400).json({
error: 'Validation error',
details: error.details.map(d => d.message),
});
}
req.body = value;
next();
};
}
app.post('/api/findings', validate(createFindingSchema), findingsController.create);Request size limits
// Express — limit JSON body
app.use(express.json({ limit: '1mb' }));
// Specific per route
app.post('/api/upload', express.json({ limit: '10mb' }), uploadHandler);Content-Type enforcement
from flask import request, abort
@app.before_request
def enforce_content_type():
if request.method in ('POST', 'PUT', 'PATCH'):
if not request.content_type or 'application/json' not in request.content_type:
abort(415, description="Content-Type must be application/json")GraphQL-specific security
GraphQL gives clients the freedom to request exactly what they need. That flexibility is also the greatest security risk: without restrictions a client can drain the entire database with a single query.
Disabling introspection in production
GraphQL introspection exposes your complete schema: types, fields, relationships, mutations. In production this is a roadmap for attackers.
# Strawberry (Python)
import strawberry
from strawberry.extensions import DisableIntrospection
schema = strawberry.Schema(
query=Query,
mutation=Mutation,
extensions=[DisableIntrospection()], # No __schema queries
)// Apollo Server (Node.js)
const { ApolloServer } = require('@apollo/server');
const server = new ApolloServer({
typeDefs,
resolvers,
introspection: process.env.NODE_ENV !== 'production',
});Query depth limiting
Prevent deeply nested queries that overload the server:
# Attack: exponentially nested query
{
user(id: 1) {
friends {
friends {
friends {
friends {
# ... until the server crashes
}
}
}
}
}
}// Apollo Server — graphql-depth-limit
const depthLimit = require('graphql-depth-limit');
const server = new ApolloServer({
typeDefs,
resolvers,
validationRules: [depthLimit(5)], // Maximum 5 levels deep
});# Strawberry — QueryDepthLimiter
from strawberry.extensions import QueryDepthLimiter
schema = strawberry.Schema(
query=Query,
extensions=[
QueryDepthLimiter(max_depth=5),
],
)Query complexity analysis
Assign costs to fields and limit the total query cost:
// Apollo Server — graphql-query-complexity
const {
getComplexity,
simpleEstimator,
fieldExtensionsEstimator,
} = require('graphql-query-complexity');
const server = new ApolloServer({
typeDefs,
resolvers,
plugins: [{
requestDidStart: () => ({
didResolveOperation({ request, document }) {
const complexity = getComplexity({
schema: server.schema,
operationName: request.operationName,
query: document,
variables: request.variables,
estimators: [
fieldExtensionsEstimator(),
simpleEstimator({ defaultComplexity: 1 }),
],
});
if (complexity > 1000) {
throw new Error(
`Query too complex: ${complexity}. Maximum: 1000.`
);
}
},
}),
}],
});Controlling batching and aliasing risks
GraphQL aliasing makes it possible to execute the same query multiple times in a single request — an effective bypass of rate limiting:
# Brute-force via aliasing — 100 login-pogingen in 1 request
{
a1: login(username: "admin", password: "password1") { token }
a2: login(username: "admin", password: "password2") { token }
a3: login(username: "admin", password: "password3") { token }
# ... 97 meer
}Solutions:
// Beperk het aantal aliases per query
const { ApolloArmor } = require('@escape.tech/graphql-armor');
const armor = new ApolloArmor({
maxAliases: { n: 10 }, // Max 10 aliases per query
maxDirectives: { n: 5 }, // Max 5 directives per query
maxDepth: { n: 5 }, // Max diepte 5
costLimit: { maxCost: 1000 },
});
const server = new ApolloServer({
typeDefs,
resolvers,
...armor.protect(),
});Field-level authorization in GraphQL
# Strawberry — permissie per veld
import strawberry
from strawberry.permission import BasePermission
from strawberry.types import Info
class IsAdmin(BasePermission):
message = "Admin-rechten vereist"
def has_permission(self, source, info: Info, **kwargs) -> bool:
return info.context.user.role == "admin"
@strawberry.type
class User:
id: int
name: str
email: str
@strawberry.field(permission_classes=[IsAdmin])
def last_login_ip(self) -> str:
return self._last_login_ipAPI versioning and deprecation
Why versioning
Breaking changes in an API without versioning break all existing clients at once. Versioning gives clients time to migrate.
Strategies
| Strategy | Example | Advantages | Disadvantages |
|---|---|---|---|
| URL-based | /api/v1/users |
Simple, explicit, cacheable | URL pollution, harder to route |
| Header-based | Accept: application/vnd.api+json;version=2 |
Clean URLs | Less visible, harder to test |
| Query parameter | /api/users?version=2 |
Easy to add | Pollutes caching, not RESTful |
Recommendation: URL-based versioning for public APIs (clear and debuggable), header-based for internal APIs (cleaner contracts).
# Flask — URL-based versioning
from flask import Blueprint
v1 = Blueprint('v1', __name__, url_prefix='/api/v1')
v2 = Blueprint('v2', __name__, url_prefix='/api/v2')
@v1.get('/users/<int:user_id>')
def get_user_v1(user_id):
user = User.query.get_or_404(user_id)
return jsonify({"id": user.id, "name": user.name})
@v2.get('/users/<int:user_id>')
def get_user_v2(user_id):
user = User.query.get_or_404(user_id)
return jsonify({
"id": user.id,
"name": user.name,
"email": user.email,
"created_at": user.created_at.isoformat(),
})Communicating deprecation
Use the Deprecation and Sunset
headers:
HTTP/1.1 200 OK
Deprecation: true
Sunset: Sat, 01 Nov 2025 00:00:00 GMT
Link: </api/v2/users>; rel="successor-version"
Logging and monitoring
What to log
| Event | Priority | Example |
|---|---|---|
| Authentication failures | High | Wrong password, invalid token, expired JWT |
| Authorization failures | High | 403 Forbidden, BOLA attempt |
| Rate limit hits | Medium | 429 Too Many Requests |
| Unexpected input | Medium | Schema validation errors, unknown fields |
| Large responses | Medium | Response > 1 MB, query returns > 1000 records |
| Admin actions | High | Create/delete user, change permissions |
| Unusual patterns | Medium | Sequential ID enumeration, bulk data export |
Structured logging
import structlog
import time
logger = structlog.get_logger()
@app.before_request
def log_request_start():
request._start_time = time.monotonic()
@app.after_request
def log_request(response):
duration = time.monotonic() - getattr(request, '_start_time', 0)
logger.info(
"api_request",
method=request.method,
path=request.path,
status=response.status_code,
duration_ms=round(duration * 1000, 2),
remote_addr=request.remote_addr,
user_id=getattr(current_user, 'id', None),
content_length=request.content_length,
)
return response// Node.js — pino structured logging
const pino = require('pino');
const logger = pino({ level: 'info' });
app.use((req, res, next) => {
const start = Date.now();
res.on('finish', () => {
logger.info({
method: req.method,
path: req.path,
status: res.statusCode,
duration_ms: Date.now() - start,
remote_addr: req.ip,
user_id: req.user?.id,
}, 'api_request');
});
next();
});Never log: passwords, tokens, session IDs, credit card numbers, national ID numbers, or other sensitive data. Sanitize request bodies before logging.
Common mistakes
| Mistake | Risk | Solution |
|---|---|---|
| Verbose error messages in production | Stack traces leak internal architecture | Generic error messages for clients, details only in server logs |
| No rate limiting | Brute-force, credential stuffing, DoS | Flask-Limiter / express-rate-limit with Redis backend |
Wildcard CORS (Access-Control-Allow-Origin: *) |
Any website can call your API | Explicit allowlist of origins |
| No input validation | Injection, business logic errors, data corruption | Pydantic / Joi / JSON Schema on every endpoint |
| Excessive data exposure | Sensitive fields (password hash, internal IDs) in response | Explicit serializers per endpoint and per role |
| Broken authentication | Token not validated, alg: none accepted |
Enforce algorithm, validate
exp/iss/sub claims |
| No object-level authorization (BOLA) | User A views user B's data via ID manipulation | Ownership check on every object endpoint |
| Mass assignment | Client sends role: admin in JSON body |
Allowlist of writable fields, ignore unknown fields |
| GraphQL introspection in production | Full schema visible to attackers | introspection: false in production |
| No TLS on internal APIs | Traffic interceptable on internal network | TLS everywhere, including internally (zero-trust) |
Checklist
| Measure | Description | Priority |
|---|---|---|
| Authentication on every endpoint | No anonymous access unless explicitly intended | Critical |
| Object-level authorization | Ownership check on every data operation | Critical |
| Input validation with schema | Pydantic, Joi, or JSON Schema on every route | Critical |
| Rate limiting | Per IP and per user, stricter on auth endpoints | Critical |
| Content-Type enforcement | Reject requests without correct Content-Type header |
High |
| Request size limits | MAX_CONTENT_LENGTH or
express.json({ limit }) |
High |
| Explicit CORS origins | No wildcards, only known domains | High |
| JWT algorithm enforcement | Explicit algorithms=[...], never
none |
Critical |
| GraphQL depth/complexity limits | Set maximum depth and query cost | High |
| Introspection off in production | Schema not visible to outsiders | High |
| Structured logging | Log all auth events, rate limits, and errors | High |
| API versioning | URL-based or header-based, deprecation headers | Medium |
| Short token lifetime | JWT exp maximum 15 minutes, refresh via separate
token |
High |
| No sensitive data in responses | Explicit serializers, field-level filtering per role | High |
| Keep error messages generic | No stack traces, no SQL error messages to clients | High |
Everyone builds APIs these days. API-first, microservices, serverless, headless CMS, BFF-pattern — the architecture diagrams are beautiful. A hundred services, a thousand endpoints, ten thousand arrows on a Miro board so complex that you need a PhD in graph theory to understand which service talks to which.
And then you ask: "How is authentication handled between service A and service B?" Silence. A long, uncomfortable silence. Followed by: "That's an internal endpoint, it doesn't need it." Ah yes. Internal. Behind the firewall. In the same Kubernetes cluster. Where nothing can ever go wrong. Where no container ever gets compromised. Where lateral movement is a myth that only exists in security training.
The pattern is always the same. First they build the API. Then they build the frontend. Then they launch. Then, when there's time left over (there's never time left over), they look at security. Rate limiting? "It's on the backlog." Input validation? "The frontend already validates it." Object-level authorization? "We trust our users." Disabling GraphQL introspection? "Oh, you can do that?"
My favourite discovery remains the GraphQL API of a
fintech startup that had full introspection enabled, no rate
limiting, no query depth limit, and a user type with
the field passwordHash. Query complexity: zero.
Security budget: also zero. But they did have a fantastic
onboarding flow and a Series A of eight million.
Summary
Secure APIs at every layer: authenticate every request with OAuth 2.0 or securely configured JWTs, authorize at object, function, and field level, validate all input against a schema (Pydantic, Joi), and limit abuse with rate limiting. For GraphQL: disable introspection in production, limit query depth and complexity, and prevent aliasing attacks. Versioning and structured logging make your API manageable and auditable.
In the next chapter we cover file uploads: how do you accept files from users without opening your server up to malware, path traversal, and denial-of-service?
Further reading in the knowledge base
These articles in the portal give you more background and practical context:
- APIs — the invisible glue of the internet
- SSL/TLS — why that padlock in your browser matters
- Encryption — the art of making things unreadable
- Password hashing — how websites store your password
- Penetration testing vs. vulnerability scans
You need an account to access the knowledge base. Log in or register.
Related security measures
These articles provide additional context and depth: