InventoryAllocation

Implementing OAuth2 for PMS API Access

Static API keys are the wrong credential for an automated rate parity pipeline: they never expire, they carry every scope the integration was ever granted, and a single leaked key lets a forgotten cron job overwrite live prices for weeks before anyone notices. This page is the build guide for replacing them with the OAuth2 client-credentials flow — the machine-to-machine grant that mints a short-lived, scope-checked bearer token for every PMS API call your parity engine makes. It is the initial grant that sits under Security & Authentication Boundaries, the gate every outbound mutation crosses before it is paced and dispatched to an OTA.

Prerequisites & environment

The token manager below runs in-process alongside the parity dispatcher; it owns the OAuth2 exchange so the rate-push logic never touches a raw credential. Pin these versions so the async and caching behaviour is reproducible:

On the access side you need an OAuth2 client registered in the PMS developer portal, its client_id / client_secret loaded from a secrets manager (HashiCorp Vault, AWS Secrets Manager, or Kubernetes secrets — never version control), the provider’s token endpoint URL, and an explicit list of granted scopes. This flow issues only an access_token; if your provider also returns a refresh_token, rotating it across a portfolio is covered separately in automating channel manager token renewal.

The client-credentials exchange and the mid-batch 401 refresh A sequence diagram across three lifelines: the Parity Engine, the PMS Token Endpoint, and the PMS Rate API. The engine POSTs grant_type=client_credentials with its client_id, client_secret and scope to the token endpoint, which returns a 200 with an access_token and expires_in. The engine caches the token with a TTL of expires_in minus 60 seconds, then sends a scoped rate-batch push to the Rate API carrying a Bearer token. When that push comes back 401 Unauthorized because the token expired, the auth-aware retry fires once: it invalidates the cache, re-exchanges for a fresh access_token, and re-attempts the push, which now returns 202 Accepted with a job id handed to an async poller. auth-aware retry — re-mints the token and re-attempts once on 401 Parity Engine PMS Token Endpoint PMS Rate API POST /token — grant_type=client_credentials client_id · client_secret · scope 200 — access_token + expires_in cache token — TTL = expires_in − 60s POST /rates/batch — Authorization: Bearer {token} 401 Unauthorized — access token expired token_manager.invalidate() POST /token — re-exchange (once) 200 — fresh access_token retry — Authorization: Bearer {fresh} 202 Accepted — job_id → async poller 1 2 3 4 5 6 7 8 9 10

Step-by-step implementation

The implementation is four parts: a scoped grant registration, a TTL-buffered token manager, an auth-aware retry wrapper, and an idempotent dispatch call.

Step 1 — Register the grant and pin least-privilege scopes

Use the Client Credentials grant for headless parity pushes — it has no user session, so it survives batch execution across hundreds of properties. Reserve Authorization Code with PKCE for the dashboard where a human overrides a rate. When you register, request the narrowest scopes the workflow actually needs and reject anything wider at config-load time.

python
from pydantic import BaseModel, field_validator

# Wildcard grants (admin:all, *) are refused outright: an over-permissioned
# token that leaks can rewrite plans on a sister property, and it fails the
# least-privilege check every hospitality PCI/GDPR audit looks for.
FORBIDDEN_SCOPES = {"*", "admin:all", "admin:*"}

class GrantConfig(BaseModel):
    client_id: str
    client_secret: str
    token_url: str
    scopes: list[str]  # e.g. ["rates:read", "inventory:write", "parities:audit"]

    @field_validator("scopes")
    @classmethod
    def reject_wildcards(cls, v: list[str]) -> list[str]:
        offending = set(v) & FORBIDDEN_SCOPES
        if offending or not v:
            raise ValueError(f"refusing wildcard/empty scope grant: {offending or 'empty'}")
        return v

    @property
    def scope_str(self) -> str:
        return " ".join(self.scopes)

Validating scopes in the model rather than at the call site is the non-obvious choice: a wildcard grant that slips through config is a silent liability that only surfaces during an incident review, so the pipeline refuses to start with one rather than deferring the failure to production.

Step 2 — Build a token manager with a refresh buffer

The token acquisition endpoint takes a POST with grant_type=client_credentials and returns access_token, expires_in, and token_type. Cache the token in-process (or in Redis for a multi-worker fleet) with a TTL set to expires_in − 60 seconds. That 60-second buffer is what prevents a 401 mid-batch when network latency or OTA rate limiting pushes a request past the token’s hard expiry.

python
import time
import hashlib
import httpx
import structlog

log = structlog.get_logger()
REFRESH_BUFFER_SECONDS = 60

class PMSOAuth2Manager:
    def __init__(self, cfg: GrantConfig):
        self.cfg = cfg
        self._access_token: str | None = None
        self._cache_expiry: float = 0.0
        self._client = httpx.AsyncClient(timeout=httpx.Timeout(connect=5.0, read=10.0, write=10.0, pool=10.0))

    async def _fetch_token(self) -> str:
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.cfg.client_id,
            "client_secret": self.cfg.client_secret,
            "scope": self.cfg.scope_str,
        }
        try:
            resp = await self._client.post(self.cfg.token_url, data=payload)
            resp.raise_for_status()
            data = resp.json()
            # Never trust the token to outlive its stated window: subtract a buffer,
            # and floor at 30s so a pathologically short expires_in can't thrash.
            ttl = max(data["expires_in"] - REFRESH_BUFFER_SECONDS, 30)
            self._access_token = data["access_token"]
            self._cache_expiry = time.time() + ttl
            log.info("oauth_token_acquired", scope=self.cfg.scope_str, ttl=ttl)
            return self._access_token
        except httpx.HTTPStatusError as exc:
            # Log a hash of the request body, never the body — it holds the secret.
            body_hash = hashlib.sha256(str(exc.request.content).encode()).hexdigest()[:12]
            log.error("oauth_token_fetch_failed", status_code=exc.response.status_code,
                      body_hash=body_hash, detail=exc.response.text)
            raise RuntimeError(f"OAuth2 acquisition failed: {exc.response.status_code}") from exc

    def invalidate(self) -> None:
        self._access_token = None
        self._cache_expiry = 0.0

    async def get_valid_token(self) -> str:
        if self._access_token and time.time() < self._cache_expiry:
            return self._access_token
        return await self._fetch_token()

Hashing the request body before logging a failure — rather than emitting detail alongside the payload — is the detail that keeps a leaked-credential incident out of your log aggregator: the hash lets you correlate repeated failures without ever writing client_secret to disk.

Step 3 — Wrap parity calls in an auth-aware retry

Any long-running batch can outlive a token even with the buffer, so wrap each call in a retry that treats a 401 specially: invalidate the cache, mint a fresh token, and re-attempt once with capped exponential backoff. This is deliberately hand-written rather than delegated to tenacity because the retry must force a credential refresh between attempts — a plain backoff would replay the same expired token. The retryable-versus-terminal split here mirrors categorizing 4xx vs 5xx sync errors.

python
import asyncio
from functools import wraps

def retry_on_auth_failure(max_retries: int = 3, backoff_factor: float = 1.5):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            token_manager: PMSOAuth2Manager = kwargs["token_manager"]
            for attempt in range(max_retries):
                token = await token_manager.get_valid_token()
                kwargs["headers"]["Authorization"] = f"Bearer {token}"
                try:
                    return await func(*args, **kwargs)
                except httpx.HTTPStatusError as exc:
                    # Only 401 is worth a token refresh; a 403 is a scope problem
                    # that no amount of retrying fixes, so let it propagate.
                    if exc.response.status_code == 401 and attempt < max_retries - 1:
                        log.warning("auth_retry_triggered", attempt=attempt + 1,
                                    endpoint=str(exc.request.url))
                        token_manager.invalidate()
                        # await asyncio.sleep, never time.sleep — the latter blocks
                        # the event loop and stalls every other in-flight push.
                        await asyncio.sleep(backoff_factor ** attempt)
                        continue
                    raise
        return wrapper
    return decorator

Distinguishing 401 (retry with a fresh token) from 403 (a scope the grant never had — let it fail loudly) is the design choice that matters most here: retrying a 403 just burns your rate-limit budget on a request that can never succeed and hides the real fix, which is widening the scope in Step 1.

Step 4 — Dispatch an idempotent parity batch

PMS rate endpoints often process updates asynchronously, so a duplicate submission without an idempotency key can trigger a double rate push — and an overbooking or parity violation downstream. Attach an idempotency key derived from the batch content so a safe retry is a no-op on the server. When the PMS answers 202 Accepted with a job ID, hand off to an async poller rather than blocking.

python
async def push_parity_batch(
    client: httpx.AsyncClient,
    token_manager: PMSOAuth2Manager,
    rate_updates: list[dict],  # each: {"property_id", "room_type_code", "rate_plan_code", "amount"}
    idempotency_key: str,
) -> dict:
    headers = {"Content-Type": "application/json", "Idempotency-Key": idempotency_key}

    @retry_on_auth_failure(max_retries=3)
    async def _execute(**kwargs) -> dict:
        resp = await client.post(
            "https://api.pms-provider.com/v2/rates/batch",
            json={"rates": rate_updates},
            headers=kwargs["headers"],
        )
        resp.raise_for_status()
        return resp.json()

    result = await _execute(token_manager=token_manager, headers=headers)
    if result.get("status") == "accepted":
        log.info("parity_job_submitted", job_id=result.get("job_id"),
                 record_count=len(rate_updates), idempotency_key=idempotency_key)
    return result

Passing token_manager and headers through kwargs is what lets the decorator rewrite the Authorization header on each attempt — the inner function reads the header the decorator just refreshed, so a retry after a 401 carries the new token rather than the stale one it failed with.

Gotchas & production notes

Verification snippet

Prove the scope guard and the TTL-buffer arithmetic before wiring the manager to a live endpoint. This asserts a wildcard grant is refused, and that the cached token is treated as expired the moment it crosses the buffered boundary.

python
import pytest

def test_wildcard_scope_is_rejected() -> None:
    with pytest.raises(ValueError):
        GrantConfig(client_id="cid", client_secret="s",
                    token_url="https://auth.example/token", scopes=["*"])

def test_ttl_applies_the_refresh_buffer() -> None:
    cfg = GrantConfig(client_id="cid", client_secret="s",
                      token_url="https://auth.example/token",
                      scopes=["rates:read", "inventory:write"])
    mgr = PMSOAuth2Manager(cfg)
    mgr._access_token = "cached"
    # expires_in was 3600; buffered TTL must retire the token 60s early.
    mgr._cache_expiry = time.time() + (3600 - REFRESH_BUFFER_SECONDS)
    assert time.time() < mgr._cache_expiry           # still valid now
    mgr._cache_expiry = time.time() - 1              # simulate crossing the boundary
    assert time.time() >= mgr._cache_expiry          # forces a re-fetch

test_wildcard_scope_is_rejected()
test_ttl_applies_the_refresh_buffer()

Asserting on the buffered boundary rather than the raw expires_in is the check that matters: an off-by-one that drops the 60-second buffer is invisible in a smoke test but shows up in production as intermittent 401s on the last property of every large batch. In a fuller suite, also mock a 401 to confirm retry_on_auth_failure calls invalidate() and re-fetches exactly once.

FAQ

Client Credentials or Authorization Code with PKCE — which grant should a rate-push pipeline use?

Client Credentials. A parity dispatcher is a headless service with no human in the loop, so it needs a machine-to-machine grant that survives an unattended batch across hundreds of properties. Authorization Code with PKCE is the right flow only for the interactive dashboard where a revenue manager signs in to override a rate — it mints a token bound to that person’s session, which is exactly what you do not want a cron job depending on.

Where should the access token live in a multi-worker fleet?

Move the cache out of process into Redis (or another shared store) keyed by client_id plus scope_str, with the TTL set to the same expires_in − 60 buffer. In-process caching means every worker independently hammers the token endpoint on cold start, and on a provider that counts token requests against your rate budget that stampede can itself trigger a 429. A shared cache lets one exchange serve the whole fleet; guard the refresh with a short lock so concurrent misses do not all re-mint at once.

My provider returned a refresh_token with a client-credentials grant — should I use it?

Usually no. The client-credentials flow is designed to re-exchange client_id / client_secret for a fresh access_token whenever the cached one expires, so a refresh_token is redundant and just widens what a leak exposes. Prefer re-running the grant. If your provider rate-limits the token endpoint hard enough that re-exchange is impractical, treat the refresh_token as a rotating secret and manage it with the pre-flight-and-lock design in OAuth2 token refresh strategies rather than the retry loop here.

Why do I get a 403 after switching PMS vendors when the code is unchanged?

Scope strings are vendor-specific and are not portable. A grant that one provider accepts as rates:write another may expect as rate_plans:mutate, and requesting the wrong name yields a 403 that is indistinguishable from a genuine permission gap. Realign the strings in Step 1 with the new vendor’s scope catalogue; because the retry wrapper deliberately never retries a 403, the failure surfaces loudly at the first push instead of quietly burning your rate budget.

← Back to Security & Authentication Boundaries