Platform
ScaiWave ScaiGrid ScaiCore ScaiBot ScaiDrive ScaiKey Models Tools & Services
Solutions
Organisations Developers Internet Service Providers Managed Service Providers AI-in-a-Box
Resources
Support Documentation Blog Downloads
Company
About Research Careers Investment Opportunities Contact
Log in

Your First Integration

A complete walk-through: take a service that reads a Salesforce credential from an environment variable and migrate it to ScaiVault — including error handling, retries, and rotation subscription.

The starting point#

python
1
2
3
4
5
6
7
import os
from salesforce_client import SalesforceClient

sf = SalesforceClient(
    client_id=os.environ["SF_CLIENT_ID"],
    client_secret=os.environ["SF_CLIENT_SECRET"],
)

This works, but:

  • Rotating the credential requires redeploying the service.
  • If the credential leaks in a log, no audit trail points at who read it and when.
  • Every service that needs Salesforce has its own copy of the secret in its own env.

We want one canonical place for the credential.

1. Store the secret in ScaiVault#

Do this once, from an admin context:

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
curl -X PUT https://scaivault.scailabs.ai/v1/secrets/integrations/salesforce/oauth \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "client_id": "3MVG9...",
      "client_secret": "REDACTED"
    },
    "secret_type": "json",
    "metadata": {
      "description": "Salesforce OAuth for all production integrations",
      "tags": ["salesforce", "oauth", "production"]
    },
    "options": {
      "max_versions": 10
    }
  }'

2. Grant your service access#

Create a service account in ScaiKey (e.g. sa:reporting-service) and bind it to a policy that lets it read this one path.

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
curl -X POST https://scaivault.scailabs.ai/v1/policies \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "reporting-service-salesforce",
    "description": "Reporting service reads Salesforce OAuth",
    "rules": [
      {
        "path_pattern": "integrations/salesforce/oauth",
        "permissions": ["read"]
      }
    ]
  }'

Bind it:

bash
1
2
3
4
5
6
7
curl -X POST https://scaivault.scailabs.ai/v1/policies/pol_abc123/bindings \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "identity_type": "service_account",
    "identity_id": "sa:reporting-service"
  }'

Now when your service authenticates as sa:reporting-service, it can read that one secret and nothing else.

3. Read the secret at runtime#

Using the Python SDK:

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import os
from scaivault_sdk import ScaiVaultClient

client = ScaiVaultClient(
    base_url="https://scaivault.scailabs.ai",
    token=os.environ["SCAIVAULT_TOKEN"],
)

secret = client.secrets.read("integrations/salesforce/oauth")
sf = SalesforceClient(
    client_id=secret.data["client_id"],
    client_secret=secret.data["client_secret"],
)

Or with plain httpx:

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import os
import httpx

resp = httpx.get(
    "https://scaivault.scailabs.ai/v1/secrets/integrations/salesforce/oauth",
    headers={"Authorization": f"Bearer {os.environ['SCAIVAULT_TOKEN']}"},
    timeout=5.0,
)
resp.raise_for_status()
secret = resp.json()
sf = SalesforceClient(
    client_id=secret["data"]["client_id"],
    client_secret=secret["data"]["client_secret"],
)
typescript
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
const resp = await fetch(
  "https://scaivault.scailabs.ai/v1/secrets/integrations/salesforce/oauth",
  { headers: { "Authorization": `Bearer ${process.env.SCAIVAULT_TOKEN}` } },
);
if (!resp.ok) throw new Error(`ScaiVault read failed: ${resp.status}`);
const secret = await resp.json();
const sf = new SalesforceClient({
  clientId: secret.data.client_id,
  clientSecret: secret.data.client_secret,
});
bash
1
2
curl -H "Authorization: Bearer $SCAIVAULT_TOKEN" \
     https://scaivault.scailabs.ai/v1/secrets/integrations/salesforce/oauth

4. Handle errors properly#

Read failures happen. Your service should distinguish between "transient — retry" and "permanent — fail loud". ScaiVault returns stable error codes for exactly this.

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import httpx
import time

def read_secret(path: str, max_attempts: int = 3) -> dict:
    for attempt in range(max_attempts):
        try:
            resp = httpx.get(
                f"https://scaivault.scailabs.ai/v1/secrets/{path}",
                headers={"Authorization": f"Bearer {token}"},
                timeout=5.0,
            )
            if resp.status_code == 200:
                return resp.json()["data"]

            err = resp.json().get("error", {})
            code = err.get("code")

            if code in ("rate_limited", "service_unavailable"):
                retry_after = int(resp.headers.get("Retry-After", 1))
                time.sleep(retry_after)
                continue

            if code in ("authentication_required", "token_expired"):
                # Token refresh is the caller's job
                raise PermissionError(f"ScaiVault auth failed: {err}")

            if code == "access_denied":
                raise PermissionError(f"ScaiVault access denied: {err}")

            raise RuntimeError(f"ScaiVault error {resp.status_code}: {err}")

        except httpx.RequestError:
            if attempt == max_attempts - 1:
                raise
            time.sleep(2 ** attempt)

    raise RuntimeError("max retries exceeded")

See Errors for the full code taxonomy.

5. Cache the secret in memory#

Hitting ScaiVault on every Salesforce call is wasteful and creates a dependency on ScaiVault availability for every inbound request. Cache the secret in process memory with a short TTL:

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
import time

_cache = {"value": None, "expires_at": 0}

def get_salesforce_creds():
    now = time.time()
    if _cache["value"] is None or now > _cache["expires_at"]:
        secret = read_secret("integrations/salesforce/oauth")
        _cache["value"] = secret
        _cache["expires_at"] = now + 300  # 5 minutes
    return _cache["value"]

Five minutes is a reasonable default: tight enough that rotations propagate within a few minutes, loose enough that you aren't hammering ScaiVault. If you need faster pickup, subscribe to events (next step) instead of polling.

6. Subscribe to rotation events#

When the secret rotates, your cache is stale. You can wait it out (up to 5 minutes in the example above) or subscribe to events for immediate invalidation.

Create a subscription:

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
curl -X POST https://scaivault.scailabs.ai/v1/subscriptions \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "reporting-service-sf-rotations",
    "paths": ["integrations/salesforce/oauth"],
    "events": ["secret.rotated", "secret.updated"],
    "delivery": {
      "type": "webhook",
      "url": "https://reporting.acme.example/scaivault/events",
      "secret": "whsec_..."
    }
  }'

In your service, add a webhook handler that invalidates the cache:

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()
WEBHOOK_SECRET = os.environ["SCAIVAULT_WEBHOOK_SECRET"].encode()

@app.post("/scaivault/events")
async def scaivault_event(request: Request):
    body = await request.body()
    sig = request.headers.get("X-ScaiVault-Signature", "")
    expected = hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        raise HTTPException(401, "bad signature")

    event = await request.json()
    if event["path"] == "integrations/salesforce/oauth":
        _cache["value"] = None  # force refresh on next read
    return {"ok": True}

Now rotation propagates to your cache within seconds. See Events and Webhooks for the full event catalog and Webhook Signatures for signature verification.

7. Wire up rotation#

Attach the secret to a rotation policy so the credential gets a new value every 90 days.

bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# Create the policy
curl -X POST https://scaivault.scailabs.ai/v1/rotation/policies \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "quarterly",
    "interval": "90d",
    "grace_period": "48h",
    "auto_generate": false
  }'

# Assign the secret to it
curl -X POST https://scaivault.scailabs.ai/v1/rotation/policies/rot_xyz/secrets \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"secret_path": "integrations/salesforce/oauth"}'

With auto_generate: false, rotation fires a rotation.due event but ScaiVault doesn't generate the new value itself — your automation picks up the event, generates a new OAuth credential in Salesforce, and writes it back with PUT /v1/secrets/.... With auto_generate: true on password-style secrets, ScaiVault generates a random string and writes the new version itself.

Done#

Your service now:

  • Reads its credential from ScaiVault at startup and on-demand.
  • Caches with a reasonable TTL.
  • Invalidates cache on rotation events.
  • Has a stable audit trail of every read.
  • Rotates automatically on a 90-day schedule with a 48-hour grace period.

If the credential leaks, you can see exactly who read it and when. If you need to rotate it right now, the POST /v1/secrets/.../rotate endpoint triggers it immediately.

What's next#

Updated 2026-05-17 13:26:50 View source (.md) rev 2