For authentication schemes not covered by built-in security types, you can implement your own validation logic in service code. This is common when integrating with systems like ServiceNow, SAP, or Keysight Hawkeye that use proprietary authentication mechanisms.
Access request headers via self.wsgi_environ using the HTTP_ prefix. For example, X-Hook-Signature becomes HTTP_X_HOOK_SIGNATURE.
ServiceNow and similar systems sign webhook payloads with HMAC-SHA256. Verify the signature before processing:
# -*- coding: utf-8 -*-
# stdlib
import hashlib
import hmac
from http import HTTPStatus
# Zato
from zato.server.service import Service
class HMACProtectedService(Service):
name = 'my.hmac.protected'
def handle(self):
# Get signature from request header
signature = self.wsgi_environ.get('HTTP_X_SIGNATURE')
if not signature:
self.response.status_code = HTTPStatus.UNAUTHORIZED
self.response.payload = {'error': 'Missing signature'}
return
# Verify the signature
secret = self.server.decrypt(self.config.api.hmac_secret)
expected = hmac.new(
secret.encode(),
self.request.raw_request.encode(),
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(expected, signature):
self.response.status_code = HTTPStatus.UNAUTHORIZED
self.response.payload = {'error': 'Invalid signature'}
return
# Signature valid - process the request
self.response.payload = {'status': 'ok'}
Test with curl:
Pass custom headers directly to the connection using the headers parameter.
For APIs using proprietary token-based authentication rather than standard OAuth 2.0, obtain and attach tokens manually. Standard OAuth 2.0 client credentials are handled automatically by Bearer token security definitions.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class CallWithCustomAuth(Service):
name = 'my.api.custom-auth'
def handle(self):
# Credentials from configuration
client_id = self.config.api.client_id
client_secret = self.server.decrypt(self.config.api.client_secret)
# Build auth request
auth_request = {
'client_id': client_id,
'client_secret': client_secret
}
# Get token from your auth provider
conn = self.out.rest['Auth Service'].conn
auth_response = conn.post(self.cid, auth_request)
token = auth_response.data['access_token']
# Use the token in your API call
headers = {'Authorization': f'Bearer {token}'}
conn = self.out.rest['Target API'].conn
response = conn.get(self.cid, headers=headers)
self.response.payload = response.data
SAP and similar systems require HMAC signatures on outgoing requests:
# -*- coding: utf-8 -*-
# stdlib
import hashlib
import hmac
import json
# Zato
from zato.server.service import Service
class CallWithHMACAuth(Service):
name = 'my.api.hmac-auth'
def handle(self):
payload = {'order_id': '12345', 'amount': 100}
payload_json = json.dumps(payload)
# Generate HMAC signature
secret = self.server.decrypt(self.config.api.hmac_secret)
signature = hmac.new(
secret.encode(),
payload_json.encode(),
hashlib.sha256
).hexdigest()
headers = {'X-Signature': signature}
conn = self.out.rest['Signed API'].conn
response = conn.post(self.cid, payload, headers=headers)
self.response.payload = response.data
Some APIs, such as Keysight Hawkeye, require explicit login before making API calls and logout afterwards. The session token obtained during login must be included in all subsequent requests.
# -*- coding: utf-8 -*-
# stdlib
from traceback import format_exc
# Zato
from zato.server.service import Service
class SessionBasedAPICall(Service):
name = 'my.api.session-based'
def handle(self):
# Credentials
username = self.config.api.username
password = self.config.api.password
# Step 1: Login to get session token
login_conn = self.out.rest['API Login'].conn
login_response = login_conn.post(self.cid, {
'username': username,
'password': password
})
session_token = login_response.data['session_token']
# Step 2: Make API calls with session token
headers = {'X-Session-Token': session_token}
try:
api_conn = self.out.rest['API Endpoint'].conn
response = api_conn.get(self.cid, headers=headers)
self.response.payload = response.data
except Exception:
self.logger.warning('API call failed, e:`%s`', format_exc())
raise
finally:
# Step 3: Always logout to release the session
logout_conn = self.out.rest['API Logout'].conn
logout_conn.post(self.cid, headers=headers)
This pattern ensures sessions are properly released even if the API call fails.