Plugin Development for API Gateways — Building Custom Extensions
In this tutorial, you'll learn about Plugin Development. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Custom plugin development allows you to extend the gateway with organization-specific functionality that built-in plugins do not cover.
What You'll Learn
By the end of this lesson, you will design plugin structure, implement lifecycle handlers, write unit tests for plugins, package and distribute plugins, and integrate them with the gateway.
Why It Matters
Built-in plugins cover common use cases, but every organization has unique requirements. Custom plugins let you implement tailored authentication, transformation, and routing logic.
Real-World Use
Durga Antivirus Pro developed a custom gateway plugin that validates scan request payloads against threat intelligence before forwarding to the scanning service.
Plugin Development Structure
flowchart TD
Plugin[Custom Plugin]
Plugin-->Handler[Handler Functions]
Plugin-->Schema[Configuration Schema]
Plugin-->Tests[Unit Tests]
Handler-->Access[access - Request Phase]
Handler-->Response[response - Response Phase]
Handler-->Rewrite[rewrite - URL Rewrite]
Schema-->Validation[Config Validation]
Tests-->Unit[Test Handlers]
Tests-->Integration[Test with Gateway]
Plugin Handler Structure
Build a custom plugin with request and response handlers.
from typing import Dict, Optional, Any, Callable
import json
import hashlib
import time
class CustomPlugin:
def __init__(self, name: str,
version: str = "1.0.0",
priority: int = 50):
self.name = name
self.version = version
self.priority = priority
self.schema = self._define_schema()
self.config: Dict = {}
def _define_schema(self) -> Dict:
return {
"type": "object",
"properties": {
"header_name": {
"type": "string",
"default": "X-Custom"
},
"enabled": {
"type": "boolean",
"default": True
}
},
"required": ["header_name"]
}
def validate_config(self, config: Dict) -> bool:
required = self.schema.get("required", [])
for field in required:
if field not in config:
print(f"Missing required config: {field}")
return False
return True
def configure(self, config: Dict):
if self.validate_config(config):
self.config = config
def access(self, request: Dict) -> Optional[Dict]:
header_name = self.config.get(
"header_name", "X-Custom"
)
if not self.config.get("enabled", True):
return None
timestamp = str(int(time.time()))
value = hashlib.sha256(
timestamp.encode()
).hexdigest()[:8]
request["headers"] = request.get("headers", {})
request["headers"][header_name] = value
return None
def response(self, request: Dict,
response: Dict) -> Optional[Dict]:
return None
def get_info(self) -> Dict:
return {
"name": self.name,
"version": self.version,
"priority": self.priority,
"schema": self.schema,
}
plugin = CustomPlugin("custom-header-injector")
plugin.configure({"header_name": "X-Request-Hash"})
error = plugin.access({
"method": "GET",
"headers": {}
})
print(f"No error: {error is None}")
info = plugin.get_info()
print(f"Plugin: {info['name']} v{info['version']}")
Plugin SDK Usage
Use a gateway SDK to interact with the request-response lifecycle.
from typing import Dict, Optional, Any
class PluginSDK:
def __init__(self):
self.hooks: Dict[str, list] = {
"init": [],
"access": [],
"response": [],
"log": [],
}
def register_hook(self, hook_name: str,
handler: Callable,
priority: int = 50):
self.hooks[hook_name].append({
"handler": handler,
"priority": priority
})
self.hooks[hook_name].sort(
key=lambda h: h["priority"]
)
def run_hooks(self, hook_name: str,
context: Dict) -> Dict:
for hook in self.hooks.get(hook_name, []):
result = hook["handler"](context)
if result:
context.update(result)
return context
def get_response_headers(self) -> Dict:
return {
"X-Powered-By": "CustomGateway",
"X-Gateway-Version": "1.0"
}
def log_error(self, message: str,
context: Optional[Dict] = None):
entry = {
"level": "error",
"message": message,
"timestamp": time.time(),
"context": context or {}
}
print(json.dumps(entry))
sdk = PluginSDK()
def log_request(context):
req = context.get("request", {})
print(f"Access: {req.get('method')} {req.get('path')}")
return {}
sdk.register_hook("access", log_request, priority=10)
context = sdk.run_hooks("access", {
"request": {"method": "GET", "path": "/api/scan"}
})
Plugin Testing
Test custom plugins with unit and integration tests.
import unittest
from typing import Dict
class TestCustomPlugin(unittest.TestCase):
def setUp(self):
from plugin_development import CustomPlugin
self.plugin = CustomPlugin("test-plugin")
self.plugin.configure({
"header_name": "X-Test-Header",
"enabled": True
})
def test_config_validation_valid(self):
valid = self.plugin.validate_config({
"header_name": "X-Valid"
})
self.assertTrue(valid)
def test_config_validation_missing_required(self):
valid = self.plugin.validate_config({})
self.assertFalse(valid)
def test_access_adds_header(self):
request = {"method": "GET", "headers": {}}
result = self.plugin.access(request)
self.assertIsNone(result)
self.assertIn("X-Test-Header", request["headers"])
def test_access_disabled_plugin(self):
self.plugin.configure({
"header_name": "X-Test",
"enabled": False
})
request = {"method": "GET", "headers": {}}
result = self.plugin.access(request)
self.assertIsNone(result)
self.assertNotIn("X-Test-Header",
request["headers"])
def test_schema_definition(self):
schema = self.plugin._define_schema()
self.assertIn("header_name",
schema["properties"])
self.assertIn("required", schema)
def test_get_info(self):
info = self.plugin.get_info()
self.assertEqual(info["name"], "test-plugin")
self.assertIn("version", info)
self.assertIn("priority", info)
if __name__ == "__main__":
unittest.main()
Plugin Distribution
Package and distribute plugins for use across environments.
from typing import Dict, Optional
import json
import zipfile
import io
import os
class PluginPackager:
def __init__(self):
self.manifest: Dict = {}
def create_manifest(self, name: str,
version: str,
description: str,
author: str,
min_gateway_version: str = "1.0"):
self.manifest = {
"name": name,
"version": version,
"description": description,
"author": author,
"min_gateway_version": min_gateway_version,
"entry": f"{name}.py",
}
def package_plugin(self, plugin_code: str,
output_path: str):
if not self.manifest:
raise ValueError(
"Create manifest first"
)
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w",
zipfile.ZIP_DEFLATED) as zf:
zf.writestr("manifest.json",
json.dumps(self.manifest))
zf.writestr(
self.manifest["entry"],
plugin_code
)
with open(output_path, "wb") as f:
f.write(buffer.getvalue())
def install_plugin(self, package_path: str,
install_dir: str):
with zipfile.ZipFile(package_path, "r") as zf:
zf.extractall(install_dir)
manifest_path = os.path.join(
install_dir, "manifest.json"
)
with open(manifest_path) as f:
return json.load(f)
packager = PluginPackager()
packager.create_manifest(
"header-injector", "1.0.0",
"Injects custom headers into requests",
"DodaTech", "2.0"
)
print(f"Manifest: {json.dumps(packager.manifest, indent=2)}")
Common Mistakes
Mistake 1: Not Handling Plugin Timeouts
A plugin that hangs blocks all gateway requests. Always set execution timeouts.
Mistake 2: Ignoring Plugin Isolation
Plugins should not modify global state. Use per-request context.
Mistake 3: No Config Validation
Invalid plugin configuration causes silent failures. Validate config schema on startup.
Mistake 4: Memory Leaks in Plugins
Plugins that accumulate data without cleanup cause memory growth. Use bounded caches.
Mistake 5: Not Logging Plugin Errors
Silent plugin failures make debugging impossible. Log all plugin errors with context.
Practice Questions
- What lifecycle hooks are available for gateway plugins?
- How do you pass configuration to a custom plugin?
- What is the purpose of a plugin schema?
- How do you test a plugin that depends on external services?
- How do you version and distribute custom plugins?
Challenge
Build a custom gateway plugin that validates incoming requests against a remote allowlist service, caches the allowlist for 60 seconds, injects X-Allowlist-Result header, and returns 403 if the request path is not in the allowlist.
FAQ
Mini Project
Build a custom gateway plugin that validates JWT tokens against a JWKS endpoint, caches public keys for 1 hour with automatic refresh, injects X-User-Id and X-User-Roles headers, and provides structured error responses for invalid or expired tokens.
What's Next
Learn about Gateway Plugins for using built-in plugins, or explore Gateway Kubernetes for deploying the gateway on Kubernetes.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro