# 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](https://zato.io/docs/dev/rest/authentication.html), 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 {#validate-incoming-requests}

Request headers are available through `self.request.http.headers`, with names lower-cased and dash-joined. For example, `X-Hook-Signature` becomes `x-hook-signature`.

### HMAC signature verification {#hmac-signature-verification}

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

```python
# -*- 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':

        # A request with a signature header has it verified ..
        if signature := self.request.http.headers.get('x-signature'):
            is_valid = self.verify_signature(signature)

        # .. and a request without one is never valid ..
        else:
            is_valid = False

        # .. anything that is not valid is rejected before any processing ..
        if not is_valid:
            self.response.status_code = HTTPStatus.UNAUTHORIZED
            self.response.payload = {'error': 'Invalid or missing signature'}
            return

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

    def verify_signature(self, signature) -> 'bool':

        # 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()

        return hmac.compare_digest(expected, signature)
```

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

```bash
# 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 {#authenticate-outgoing-calls}

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

### Custom token header {#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](https://zato.io/docs/dev/rest/authentication.html) security definitions.

```python
# -*- 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(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(headers=headers)

        self.response.payload = response.data
```

### HMAC-signed requests {#hmac-signed-requests}

SAP and similar systems require HMAC signatures on outgoing requests:

```python
# -*- 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 sent in the request body
        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(payload, headers=headers)

        self.response.payload = response.data
```

### Session-based login and logout {#session-based-login-and-logout}

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:

```python
# -*- 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({
            '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(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(headers=headers)
```

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

## See also {#see-also}

- [Authentication](https://zato.io/docs/dev/rest/authentication.html) - The built-in security types to check before implementing your own
- [Webhooks](https://zato.io/docs/dev/rest/webhooks.html) - Receiving and verifying signed events from external systems
- [Calling REST APIs](https://zato.io/docs/dev/rest/calling-apis.html) - The headers parameter and everything else a call accepts

## Learn more {#learn-more}

- [REST integration overview](https://zato.io/integrations/rest/) - Start here to see how teams build REST integrations
- [In-depth API tutorial](https://zato.io/tutorials/main/01.html) - Master architectural patterns for production-ready integrations
- [Modern REST API tutorial](https://zato.io/tutorials/rest-api/python.html) - Design, build and secure REST APIs with Python