计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
podium-auth【效率与协作】Authenticate production Podium integrations and survive the auth-side failures — OAuth2 access-token expiry storms, refresh-token decay after 90 days of non-use, scope drift on re-grant, secret rotation without downtime, multi-tenant token routing, leakage in commits. Use when hardening token caching, building a refresh-token decay monitor, rotating …
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: podium-auth description: Authenticate production Podium integrations and survive the auth-side failures — OAuth2 access-token expiry storms, refresh-token decay after 90 days of non-use, scope drift on re-grant, secret rotation without downtime, multi-tenant token routing, leakage in commits. Use when hardening token caching, building a refresh-token decay monitor, rotating Podium client credentials, or recovering from 401/403 auth cascades. Trigger with "podium auth", "podium oauth", "podium token refresh", "podium scope drift", "podium credential rotation", "podium multi-location auth". allowed-tools: Read, Write, Edit, Bash(curl:), Bash(jq:), Bash(python3:), Bash(openssl:), Grep version: 2.12.0 license: MIT author: Jeremy Longshore jeremy@intentsolutions.io compatibility: Designed for Claude Code tags:
Authenticate a service to the Podium API and operate the auth layer in production. This is not a setup walkthrough — it is the auth code your integration runs at 3am when a refresh token expires after the long weekend, when a Podium admin removes a scope on re-grant, when an agency credential router sends a request to the wrong location, and when on-call needs to rotate a leaked client secret without dropping in-flight call-transcript webhooks.
The six production failures this skill prevents:
401 is wrong.403 on previously-working endpoints. Retrying does not help..env hygiene is non-optional.client_id, client_secret, redirect_uri from the app's OAuth tabrefresh_tokenBuild in this order. Each section neutralizes one production failure mode.
Cache the access token in-process keyed by organization, and refresh proactively at 80% of TTL behind a single-flight lock so concurrent callers serialize on one refresh.
import asyncio
import time
from dataclasses import dataclass
from typing import Optional
import httpx
@dataclass
class CachedToken:
value: str
expires_at: float # unix seconds
class PodiumAuth:
TOKEN_URL = "https://accounts.podium.com/oauth/token"
def __init__(self, client_id: str, client_secret: str, refresh_token: str):
self.client_id = client_id
self.client_secret = client_secret
self.refresh_token = refresh_token
self._cached: Optional[CachedToken] = None
self._lock = asyncio.Lock()
async def get_token(self) -> str:
# Refresh at 80% of TTL — token endpoint can throttle if every call refreshes
if self._cached and time.time() < self._cached.expires_at - 600:
return self._cached.value
async with self._lock:
# Re-check inside the lock — another coroutine may have refreshed
if self._cached and time.time() < self._cached.expires_at - 600:
return self._cached.value
await self._refresh()
return self._cached.value
async def _refresh(self) -> None:
async with httpx.AsyncClient(timeout=10) as c:
r = await c.post(
self.TOKEN_URL,
data={
"grant_type": "refresh_token",
"refresh_token": self.refresh_token,
"client_id": self.client_id,
"client_secret": self.client_secret,
},
)
if r.status_code != 200:
raise PodiumAuthError(r.status_code, r.text)
body = r.json()
self._cached = CachedToken(
value=body["access_token"],
expires_at=time.time() + body["expires_in"],
)
# Podium rotates the refresh token on every refresh — persist the new one
if "refresh_token" in body:
self.refresh_token = body["refresh_token"]
await self._persist_refresh_token(body["refresh_token"])
class PodiumAuthError(Exception):
def __init__(self, status: int, body: str):
super().__init__(f"Podium auth failed {status}: {body}")
self.status = status
self.body = body
The single-flight lock is non-negotiable. Under burst load (a Shopify webhook fans out 200 review requests at midnight when the access token has just expired), every request races to the token endpoint, Podium throttles, and the burst fails atomically.
Podium rotates the refresh token on every successful refresh. The old refresh token is invalidated immediately. If your process refreshes successfully but crashes before persisting the new refresh token, the next process startup has a dead credential.
Persist the new refresh token to your secret store inside the refresh call, before returning the new access token to the caller:
async def _persist_refresh_token(self, new_refresh: str) -> None:
# Replace with your secret store: AWS Secrets Manager, GCP Secret Manager, SOPS, etc.
# Atomic write — temp file + rename, never a partial write.
import os, tempfile, json
path = os.environ["PODIUM_REFRESH_TOKEN_FILE"]
fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path), prefix=".podium_refresh.")
try:
with os.fdopen(fd, "w") as f:
json.dump({"refresh_token": new_refresh, "rotated_at": time.time()}, f)
os.replace(tmp, path)
except Exception:
os.unlink(tmp)
raise
Podium refresh tokens die after 90 days of non-use. Track last_used_at alongside the token; warn at day 60, page at day 75, hard-fail at day 85 with instructions for re-authorization.
DECAY_WARN_DAYS = 60
DECAY_PAGE_DAYS = 75
DECAY_HARD_FAIL_DAYS = 85
def check_decay(last_used_at: float) -> None:
age_days = (time.time() - last_used_at) / 86400
if age_days >= DECAY_HARD_FAIL_DAYS:
raise PodiumAuthError(
0,
f"Refresh token unused for {age_days:.0f}d (>{DECAY_HARD_FAIL_DAYS}d) — "
"user must re-authorize the Podium OAuth app before requests resume.",
)
if age_days >= DECAY_PAGE_DAYS:
page_oncall(
f"Podium refresh token nearing expiry: {age_days:.0f}d / 90d",
severity="high",
)
elif age_days >= DECAY_WARN_DAYS:
log_warn(f"Podium refresh token age: {age_days:.0f}d / 90d")
This protects the seasonal-business case explicitly: a campervan retailer's off-season is exactly the failure mode where naive integrations break silently and ship operators discover it when they reopen for summer.
When a Podium admin re-grants your app, the new access token's scope set is whatever the admin selected — which may be a subset of what you previously had. Validate scopes immediately after each refresh; fail loudly rather than discover the drift on a 403 in production:
REQUIRED_SCOPES = {
"conversations.read",
"conversations.write",
"contacts.read",
"contacts.write",
"reviews.read",
"reviews.write",
}
def validate_scopes(token_body: dict) -> None:
granted = set(token_body.get("scope", "").split(" "))
missing = REQUIRED_SCOPES - granted
if missing:
raise PodiumAuthError(
0,
f"Scope drift detected — missing: {sorted(missing)}. "
"A Podium org admin must re-grant these scopes in the OAuth app settings.",
)
Wire validate_scopes(body) into _refresh() immediately after the JSON parse, before assigning to self._cached.
Podium client secrets are long-lived and grant access to every endpoint the OAuth app is scoped for. Never put them in source code, log output, or git history.
# .gitignore — verify these are present
.env
.env.local
.env.*.local
podium-credentials.json
podium-refresh-token.json
# Audit the repo for accidentally-committed credentials before this becomes prod
git log --all --full-history --oneline -- .env
grep -rnE "podium.*(client_secret|refresh_token)\s*=\s*['\"]" --include="*.py" --include="*.ts" --include="*.json" .
For prod, encrypt credentials at rest with SOPS + age (Intent Solutions standard) and decrypt in-process — never write the plaintext to disk.
When rotating a leaked or aging client secret:
podium/client_secret_v2).client_id.async def verify_credential(token: str) -> bool:
async with httpx.AsyncClient(timeout=5) as c:
r = await c.get(
"https://api.podium.com/v4/me",
headers={"Authorization": f"Bearer {token}"},
)
return r.status_code in (200, 204)
| HTTP Status | Podium Error | Root Cause | Action |
|---|---|---|---|
401 Unauthorized | invalid_token | Access token expired or malformed | Refresh; if refresh also 401, re-authorize |
401 Unauthorized | invalid_grant | Refresh token expired (90d) or revoked | User must re-authorize the OAuth app |
403 Forbidden | insufficient_scope | Scope removed on re-grant | Admin re-grants required scopes |
400 Bad Request | invalid_client | Wrong client_id/secret combination | Verify against Podium dev console |
429 Too Many Requests | rate_limited | Token endpoint throttled (auth burst) | Back off with Retry-After header |
500/502/503 | server_error | Podium-side transient | Exponential backoff with jitter, max 4 attempts |
curl -s -X POST https://accounts.podium.com/oauth/token \
-d grant_type=refresh_token \
-d refresh_token="{your-refresh-token}" \
-d client_id="{your-client-id}" \
-d client_secret="{your-client-secret}" | jq '{access_token, expires_in, scope}'
auth = PodiumAuth(
client_id=os.environ["PODIUM_CLIENT_ID"],
client_secret=os.environ["PODIUM_CLIENT_SECRET"],
refresh_token=load_refresh_token(),
)
async def podium_get(path: str) -> httpx.Response:
token = await auth.get_token()
async with httpx.AsyncClient() as c:
return await c.get(
f"https://api.podium.com{path}",
headers={"Authorization": f"Bearer {token}"},
)
class PodiumOrgRouter:
def __init__(self, credentials: dict[str, dict]):
# credentials = {"acme-rv": {client_id, client_secret, refresh_token}, ...}
self._auths: dict[str, PodiumAuth] = {
org: PodiumAuth(**creds) for org, creds in credentials.items()
}
async def get_token(self, org_slug: str) -> str:
auth = self._auths.get(org_slug)
if not auth:
raise KeyError(f"No Podium credentials for org: {org_slug}")
return await auth.get_token()
.gitignore audited for credential leakage patterns