OAuth2 at the Gateway — Authorization Code Flow and Token Management
In this tutorial, you'll learn about OAuth Gateway. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
An OAuth2-enabled gateway handles the authorization code flow, validates access tokens, enforces scopes, and propagates identity to downstream services.
What You'll Learn
By the end of this lesson, you will implement OAuth2 authorization code flow at the gateway, validate and introspect access tokens, enforce scopes, and pass token claims to backend services.
Why It Matters
OAuth2 integration at the gateway provides a centralized authorization layer that supports social login, machine-to-machine auth, and fine-grained permission control.
Real-World Use
Durga Antivirus Pro uses OAuth2 at the gateway with scopes like scan:read, scan:write, and admin:reports to control access to different API endpoints based on user roles.
OAuth2 Gateway Flow
sequenceDiagram
Browser->>Gateway: GET /api/data
Gateway->>Browser: 302 to Auth Server
Browser->>Auth: Login & Consent
Auth->>Browser: 302 with Auth Code
Browser->>Gateway: GET /callback?code=abc
Gateway->>Auth: POST /token (code exchange)
Auth->>Gateway: Access Token + Refresh Token
Gateway->>Gateway: Validate Token & Extract Claims
Gateway->>Backend: Forward with User Headers
Backend->>Browser: Response
OAuth2 Token Validator
The gateway validates access tokens against the authorization server.
import requests
from typing import Dict, Optional, Tuple
from datetime import datetime
import time
class OAuth2Gateway:
def __init__(self, token_url: str,
introspection_url: str,
client_id: str, client_secret: str,
scope_map: Optional[Dict[str, list]] = None):
self.token_url = token_url
self.introspection_url = introspection_url
self.client_id = client_id
self.client_secret = client_secret
self.scope_map = scope_map or {}
self.cache: Dict[str, Tuple[bool, float, Dict]] = {}
def introspect_token(self, token: str
) -> Tuple[bool, Optional[Dict]]:
cached = self.cache.get(token)
if cached:
valid, expiry, data = cached
if time.time() < expiry:
return valid, data
response = requests.post(
self.introspection_url,
auth=(self.client_id, self.client_secret),
data={"token": token},
timeout=5
)
result = response.json()
active = result.get("active", False)
if active:
exp = result.get("exp", time.time() + 60)
self.cache[token] = (
True, min(exp, time.time() + 60), result
)
return active, result if active else None
def check_scope(self, token_data: Dict,
endpoint: str) -> bool:
token_scopes = token_data.get("scope", "").split()
required_scopes = self.scope_map.get(endpoint, [])
if not required_scopes:
return True
return any(
s in token_scopes for s in required_scopes
)
def gateway_auth(self, token: str,
endpoint: str
) -> Tuple[int, Optional[Dict]]:
valid, data = self.introspect_token(token)
if not valid:
return 401, {"error": "invalid_token"}
if not self.check_scope(data, endpoint):
return 403, {"error": "insufficient_scope"}
return 200, data
gateway_oauth = OAuth2Gateway(
"https://auth.example.com/token",
"https://auth.example.com/introspect",
"gateway-client", "secret123",
scope_map={
"/api/scan": ["scan:read", "scan:write"],
"/api/reports": ["admin:reports"]
}
)
status, data = gateway_oauth.gateway_auth(
"some-token", "/api/scan"
)
print(f"Auth status: {status}")
Authorization Code Exchange
The gateway handles the code exchange step of the authorization code flow.
import requests
from typing import Dict, Optional
from urllib.parse import urlencode
class AuthorizationCodeHandler:
def __init__(self, auth_url: str, token_url: str,
client_id: str, client_secret: str,
redirect_uri: str):
self.auth_url = auth_url
self.token_url = token_url
self.client_id = client_id
self.client_secret = client_secret
self.redirect_uri = redirect_uri
def build_authorization_url(self, state: str,
scopes: list[str]) -> str:
params = {
"response_type": "code",
"client_id": self.client_id,
"redirect_uri": self.redirect_uri,
"scope": " ".join(scopes),
"state": state,
}
return f"{self.auth_url}?{urlencode(params)}"
def exchange_code(self, code: str
) -> Optional[Dict]:
response = requests.post(
self.token_url,
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": self.redirect_uri,
"client_id": self.client_id,
"client_secret": self.client_secret,
},
timeout=5
)
if response.status_code == 200:
return response.json()
return None
def refresh_token(self, refresh_token: str
) -> Optional[Dict]:
response = requests.post(
self.token_url,
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": self.client_id,
"client_secret": self.client_secret,
},
timeout=5
)
if response.status_code == 200:
return response.json()
return None
handler = AuthorizationCodeHandler(
"https://auth.example.com/authorize",
"https://auth.example.com/token",
"gateway-client", "secret123",
"https://gateway.example.com/callback"
)
auth_url = handler.build_authorization_url(
"state-abc", ["scan:read", "scan:write"]
)
print(f"Redirect user to: {auth_url}")
Scope-Based Routing
The gateway can route requests based on token scopes.
from typing import Dict, Optional, List, Callable
class ScopeRouter:
def __init__(self):
self.routes: Dict[str, List[Dict]] = {}
def add_route(self, path: str,
required_scope: str,
backend: str):
if path not in self.routes:
self.routes[path] = []
self.routes[path].append({
"scope": required_scope,
"backend": backend
})
def route(self, path: str,
token_scopes: List[str]
) -> Optional[str]:
routes = self.routes.get(path, [])
for route in routes:
if route["scope"] in token_scopes:
return route["backend"]
if routes:
return routes[0]["backend"]
return None
router = ScopeRouter()
router.add_route("/api/reports", "admin:reports", "report-service-vip")
router.add_route("/api/reports", "report:read", "report-service")
backend = router.route("/api/reports", ["report:read"])
print(f"Routed to: {backend}")
Common Mistakes
Mistake 1: Not Validating the State Parameter
Without state validation, the gateway is vulnerable to CSRF Attacks on the callback endpoint.
Mistake 2: Caching Introspection Results Too Aggressively
Cached introspection results can allow revoked tokens to be used. Cache for seconds, not minutes.
Mistake 3: Ignoring Token Expiration in the Gateway
The gateway must check exp on every request. Do not assume the token is still valid because it was valid a minute ago.
Mistake 4: Not Handling Refresh Token Rotation
When a refresh token is used, the authorization server may return a new refresh token. The gateway must update it.
Mistake 5: Leaking Tokens in Logs
Never log access tokens or refresh tokens. Log only the token hash or a truncated form.
Practice Questions
- What is the purpose of the authorization code in OAuth2 flow?
- Why should the gateway introspect tokens rather than trusting them blindly?
- How does scope enforcement work at the gateway level?
- What is the state parameter and why is it important?
- How do you handle token refresh in a gateway context?
Challenge
Build an OAuth2 gateway plugin that supports authorization code flow, token introspection with caching (max 30 seconds), scope-based endpoint access control, and injects X-User-Id and X-Scope headers for downstream services.
FAQ
Mini Project
Build an OAuth2 gateway that redirects unauthenticated users to the authorization server, handles the callback with code exchange, validates the access token via introspection on every request, enforces scope-based access, and sets X-User-Id and X-Scope headers.
What's Next
Learn about JWT Gateway for stateless authentication, or explore API Key Gateway for machine-to-machine authentication patterns.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro