Custom authentication

HMAC signatures, proprietary tokens and session-based flows - authentication schemes you implement yourself.

For authentication schemes not covered by built-in security types, you can implement your own validation logic in service code. Systems such as ServiceNow, SAP or Keysight Hawkeye use proprietary authentication mechanisms that work this way.

Channels

Validate incoming requests

Request headers are available through self.wsgi_environ under the HTTP_ prefix. For example, X-Hook-Signature becomes HTTP_X_HOOK_SIGNATURE.

HMAC signature verification

ServiceNow and similar systems sign webhook payloads with HMAC-SHA256. The service verifies the signature before it processes the request:

# -*- coding: utf-8 -*-

# stdlib
import hashlib
import hmac
from http import HTTPStatus

# Zato
from zato.server.service import Service

class HMACProtectedService(Service):
    """ Verifies an HMAC-SHA256 signature before processing a request.
    """
    name = 'demo.rest.hmac-protected'

    def handle(self) -> 'None':

        # The signature from the request header ..
        signature = self.wsgi_environ.get('HTTP_X_SIGNATURE')

        # .. a request without one is rejected before any processing ..
        if not signature:
            self.response.status_code = HTTPStatus.UNAUTHORIZED
            self.response.payload = {'error': 'Missing signature'}
            return

        # .. the expected signature is computed over the raw request body ..
        secret = self.server.decrypt(self.config.api.hmac_secret)
        expected = hmac.new(
            secret.encode(),
            self.request.raw.encode(),
            hashlib.sha256
        ).hexdigest()

        # .. a mismatch is rejected too ..
        if not hmac.compare_digest(expected, signature):
            self.response.status_code = HTTPStatus.UNAUTHORIZED
            self.response.payload = {'error': 'Invalid signature'}
            return

        # .. and only a request with a valid signature reaches this point.
        self.response.payload = {'status': 'ok'}

To verify the flow, compute a signature over a test payload and send both:

# Generate a signature and call the endpoint
PAYLOAD='{"data":"test"}'
SIGNATURE=$(echo -n "$PAYLOAD" | openssl dgst -sha256 -hmac "your-secret" | cut -d' ' -f2)
curl -X POST http://localhost:11223/api/hmac/protected \
    -H "Content-Type: application/json" \
    -H "X-Signature: $SIGNATURE" \
    -d "$PAYLOAD"

Outgoing connections

Authenticate outgoing calls

Pass custom headers directly to the connection through the headers parameter.

Custom token header

For APIs using proprietary token-based authentication rather than standard OAuth 2.0, obtain and attach tokens yourself. 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):
    """ Obtains a proprietary token and calls an API with it.
    """
    name = 'demo.rest.custom-auth'

    def handle(self) -> 'None':

        # Credentials from configuration ..
        client_id = self.config.api.client_id
        client_secret = self.server.decrypt(self.config.api.client_secret)

        # .. the auth provider exchanges them for a token ..
        auth_request = {
            'client_id': client_id,
            'client_secret': client_secret
        }
        conn = self.rest['Auth Service']
        auth_response = conn.post(self.cid, auth_request)
        token = auth_response.data['access_token']

        # .. and the token authenticates the actual API call.
        headers = {'Authorization': f'Bearer {token}'}

        conn = self.rest['Target API']
        response = conn.get(self.cid, headers=headers)

        self.response.payload = response.data

HMAC-signed requests

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 = 'demo.rest.hmac-auth'

    def handle(self) -> 'None':

        payload = {'order_id': '12345', 'amount': 100}
        payload_json = json.dumps(payload)

        # The signature covers the exact bytes the request carries
        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.rest['Signed API']
        response = conn.post(self.cid, payload, headers=headers)

        self.response.payload = response.data

APIs such as Keysight Hawkeye require an explicit login before API calls and a logout afterwards, with the session token from the login included in every request in between:

# -*- coding: utf-8 -*-

# stdlib
from traceback import format_exc

# Zato
from zato.server.service import Service

class SessionBasedAPICall(Service):
    name = 'demo.rest.session-based'

    def handle(self) -> 'None':

        # Credentials from configuration
        username = self.config.api.username
        password = self.config.api.password

        # Step 1: Log in to receive a session token
        login_conn = self.rest['API Login']
        login_response = login_conn.post(self.cid, {
            'username': username,
            'password': password
        })
        session_token = login_response.data['session_token']

        # Step 2: Call the API with the session token
        headers = {'X-Session-Token': session_token}

        try:
            api_conn = self.rest['API Endpoint']
            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: Log out to release the session
            logout_conn = self.rest['API Logout']
            logout_conn.post(self.cid, headers=headers)

The finally block runs the logout even when the API call raises an error, so the session is always released.

Page	What it covers
Authentication	The built-in security types to check before implementing your own
Webhooks	Receiving and verifying signed events from external systems
Calling REST APIs	The headers parameter and everything else a call accepts

Learn more

REST integration overview Start here to see how teams build REST integrations

In-depth API tutorial Master architectural patterns for production-ready integrations

Modern REST API tutorial Design, build and secure REST APIs with Python