API Security and Service-to-Service Trust
Include REST, GraphQL, RPC, webhooks, message queues, internal service APIs, batch endpoints, and administrative endpoints.
10.1 API threat surface
Include REST, GraphQL, RPC, webhooks, message queues, internal service APIs, batch endpoints, and administrative endpoints.
10.2 API controls
Authenticate callers using an appropriate mechanism.
Authorize every operation and object.
Validate schema, type, range, length, and state transitions.
Limit request size, pagination, execution cost, concurrency, and frequency.
Use timeouts, quotas, and circuit breakers.
Avoid exposing internal exception messages.
Return only required fields.
Version APIs carefully and retire insecure versions.
Protect sensitive responses from caching.
Log security-relevant events with correlation IDs, not credentials.
10.3 Rate limiting
Apply rate limits according to identity, IP/network, tenant, operation, and cost where appropriate. Consider distributed deployments, burst behavior, legitimate shared networks, and bypass paths. Rate limiting should complement, not replace, abuse detection and resource limits.
A token bucket with a test clock Output (CPython 3.13.1):
Passing the time in makes the limiter deterministic to test. In production the bucket state lives in a shared store keyed by identity, tenant and operation, so every instance of the service sees the same counts.
class TokenBucket:
def __init__(self, capacity: int, refill_per_second: float):
self.capacity = capacity
self.rate = refill_per_second
self.tokens = float(capacity)
self.updated = 0.0
def allow(self, now: float) -> bool:
self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
self.updated = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
bucket = TokenBucket(capacity=3, refill_per_second=0.5) # bursts of 3, then 1 every 2 s
for t in [0, 0.1, 0.2, 0.3, 1.0, 2.4, 2.5, 6.5]:
print(f"t={t:>4}s {'allowed' if bucket.allow(t) else 'limited'}")t= 0s allowed
t= 0.1s allowed
t= 0.2s allowed
t= 0.3s limited
t= 1.0s limited
t= 2.4s allowed
t= 2.5s limited
t= 6.5s allowedratelimit10.4 Idempotency and replay
For operations such as payments or order creation, retries can duplicate side effects. Use idempotency keys or domain-level uniqueness, bind them to the caller and operation, define expiry, and perform the state transition atomically. Do not assume a client will send a request only once.
Idempotency enforced by the database Output (CPython 3.13.1):
The UNIQUE constraint makes the check and the write one atomic step, so two concurrent retries cannot both create a payment.
import sqlite3
con = sqlite3.connect(":memory:")
con.execute("""CREATE TABLE payments (
caller TEXT NOT NULL, idempotency_key TEXT NOT NULL, amount_cents INTEGER NOT NULL,
UNIQUE (caller, idempotency_key))""")
def create_payment(caller: str, key: str, amount_cents: int) -> str:
try:
with con:
con.execute("INSERT INTO payments VALUES (?, ?, ?)", (caller, key, amount_cents))
return "created"
except sqlite3.IntegrityError:
stored = con.execute("SELECT amount_cents FROM payments WHERE caller = ? AND idempotency_key = ?",
(caller, key)).fetchone()[0]
return "replayed (same result)" if stored == amount_cents else "rejected: key reused with different request"
print(create_payment("user-1", "k-001", 2500))
print(create_payment("user-1", "k-001", 2500)) # client retry after a timeout
print(create_payment("user-1", "k-001", 9900))
print(create_payment("user-2", "k-001", 2500)) # keys are scoped per caller
print("rows:", con.execute("SELECT count(*) FROM payments").fetchone()[0])created
replayed (same result)
rejected: key reused with different request
created
rows: 210.5 Webhook verification
Verify signatures using the provider's documented algorithm and canonicalization.
Compare MACs/signatures using a suitable library and timing-safe comparison where appropriate.
Check timestamp windows and replay protection.
Treat the body as untrusted even after signature verification.
Process events idempotently.
Rotate webhook secrets and protect delivery logs.
Verifying a signed webhook Output (CPython 3.13.1): The timestamp is part of the signed content, so it cannot be changed without breaking the signature; the window and the seen-ID set stop old or repeated deliveries. Providers document their own exact format — follow it rather than this sketch.
import hashlib
import hmac
import time
SECRET = b"test-only-webhook-secret" # in production: from a secrets manager
TOLERANCE = 300 # seconds
seen_ids = set()
def sign(timestamp: int, body: bytes) -> str:
return hmac.new(SECRET, f"{timestamp}.".encode() + body, hashlib.sha256).hexdigest()
def verify(event_id: str, timestamp: int, body: bytes, signature: str, now: int) -> str:
if abs(now - timestamp) > TOLERANCE:
return "rejected: outside time window"
if not hmac.compare_digest(sign(timestamp, body), signature):
return "rejected: bad signature"
if event_id in seen_ids:
return "ignored: duplicate delivery"
seen_ids.add(event_id)
return "accepted"
now = int(time.time())
body = b'{"type": "payment.succeeded", "order": "A-17"}'
sig = sign(now, body)
print(verify("evt_1", now, body, sig, now))
print(verify("evt_1", now, body, sig, now + 5))
print(verify("evt_2", now, body.replace(b"A-17", b"A-18"), sig, now))
print(verify("evt_3", now - 900, body, sign(now - 900, body), now))accepted
ignored: duplicate delivery
rejected: bad signature
rejected: outside time window10.6 Service-to-service identity
Use workload identity, short-lived credentials, mutual TLS, or a managed service identity where appropriate. Authenticate the service and separately authorize the operation. Internal network location alone is not proof of trust.
10.7 GraphQL-specific controls
Authorize at resolvers/data-access boundaries, not only at the root.
Limit query depth, complexity, aliases, batching, and response size.
Avoid revealing schema details beyond the intended policy.
Prevent N+1 query amplification.
Apply tenant scoping to every resolver.
Consider persisted/allowlisted queries for high-risk deployments.
Exercise
Design an API for course enrollment, progress updates, certificates, and administrative reporting. Define authentication, object authorization, quotas, idempotency, and negative tests.
Quiz
? A client retries `POST /payments` after a timeout. What prevents a double charge?
+ An idempotency key bound to the caller and operation, enforced atomically :: The second request returns the first result.
- Rate limiting :: It limits volume, not duplicates.
- HTTPS :: Transport security does not deduplicate.
- Returning HTTP 500 on the retry :: That does not tell whether the first attempt succeeded.
? Which checks belong in webhook verification? (select all that apply)
+ Signature over the exact body using the provider's algorithm :: Proves origin and integrity.
+ Timestamp window :: Limits replay of old deliveries.
+ Duplicate-event detection :: Providers retry; processing must be idempotent.
- Trusting the source IP alone :: IPs can change and are not proof of identity.
? Two internal services talk over a private network. What does the course recommend?
- Trust any call from inside the network. :: Network location is not proof of trust.
+ Authenticate the calling service and authorize the operation separately. :: Workload identity or mTLS plus authorization.
- Share one long-lived API key across all services. :: Hard to rotate, scope or attribute.
- Disable TLS internally fo
…