ops: manual LongPort token refresh job (dry-run default)

.github/workflows/refresh-longport-token.yml

@@ -0,0 +1,92 @@
+name: Refresh LongPort Access Token
+
+# Pre-expiry Legacy token refresh into Secret Manager.
+# Default dry_run=true (plan only). Live mode: paper → sg → hk last.
+# Never prints secret values. No schedule (manual only).
+on:
+  workflow_dispatch:
+    inputs:
+      target:
+        description: "Account target (prefer paper, then sg; hk last)."
+        required: true
+        type: choice
+        options:
+          - paper
+          - sg
+          - hk
+      dry_run:
+        description: "When true (default), only plan; do not call LongPort or write SM."
+        required: true
+        type: boolean
+        default: true
+      force:
+        description: "Refresh even when days remaining >= threshold (still blocked if expired)."
+        required: false
+        type: boolean
+        default: false
+      refresh_threshold_days:
+        description: "Refresh when days remaining is below this value."
+        required: false
+        type: string
+        default: "30"
+
+permissions:
+  contents: read
+
+concurrency:
+  group: refresh-longport-token-${{ inputs.target }}
+  cancel-in-progress: false
+
+jobs:
+  refresh:
+    if: github.repository == 'QuantStrategyLab/LongBridgePlatform' && github.ref == 'refs/heads/main'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      GCP_PROJECT_ID: longbridgequant
+      GCP_WORKLOAD_IDENTITY_PROVIDER: projects/252919773759/locations/global/workloadIdentityPools/github-actions/providers/github-main
+      GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: longbridge-platform-deploy@longbridgequant.iam.gserviceaccount.com
+      TARGET: ${{ inputs.target }}
+      DRY_RUN: ${{ inputs.dry_run }}
+      FORCE: ${{ inputs.force }}
+      REFRESH_THRESHOLD_DAYS: ${{ inputs.refresh_threshold_days }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v5
+
+      - name: Authenticate to Google Cloud (deploy SA)
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: python -m pip install --quiet 'google-cloud-secret-manager>=2.20' 'requests>=2.31'
+
+      - name: Refresh or dry-run plan (names + days only)
+        env:
+          PYTHONPATH: ${{ github.workspace }}
+        run: |
+          set -euo pipefail
+          args=(
+            --project "$GCP_PROJECT_ID"
+            --target "$TARGET"
+            --refresh-threshold-days "$REFRESH_THRESHOLD_DAYS"
+          )
+          if [[ "${DRY_RUN}" == "true" ]]; then
+            args+=(--dry-run)
+          else
+            args+=(--no-dry-run)
+          fi
+          if [[ "${FORCE}" == "true" ]]; then
+            args+=(--force)
+          fi
+          python scripts/refresh_longport_access_token.py "${args[@]}" | tee "$GITHUB_STEP_SUMMARY"

docs/longport_token_refresh_runbook.md

@@ -68,7 +68,7 @@ Treat OAuth as a **follow-on migration** after path A is green, not the emergenc
 | --- | --- | --- | --- |
 | A0 (this PR) | Runbook + read-only JWT expiry inspector (no refresh, no SM write) | paper / hk / sg | Docs + optional manual inspect only |
 | A1 | QPK: send `expired_at`; fail closed on pre-expiry refresh failure; keep SM write via store_rw | kit first | Shared kit; pin bump in platform after |
-| A2 | Dedicated refresh job (Actions `workflow_dispatch` → then schedule) calling refresh **before** expiry, independent of `RUNTIME_TARGET_ENABLED` | **paper first**, then **sg**, **hk last** | Mutates token SM version; invalidates previous Access Token on success |
+| A2 | Dedicated refresh job ([`refresh-longport-token.yml`](../.github/workflows/refresh-longport-token.yml), default `dry_run=true`) calling refresh **before** expiry, independent of `RUNTIME_TARGET_ENABLED`; schedule only after paper/sg live succeed | **paper first**, then **sg**, **hk last** | Mutates token SM version when `dry_run=false`; invalidates previous Access Token on success |
 | A3 | Alert when days-to-exp < threshold (reuse inspector) | all | Notify only |
 | B (later) | OAuth spike on paper only, SDK 4.x, SM-backed refresh token | paper → sg → hk | Large: auth model + SDK + deploy |
 
@@ -104,3 +104,4 @@ Constraints this plan respects: no `independent_get` / ingress changes, no produ
 - Probe history branch (no refresh): `main.py` `run_probe`
 - Rotate workflow: `.github/workflows/rotate-longport-secrets.yml`
 - Read-only expiry inspect: `.github/workflows/inspect-longport-token-expiry.yml` + `scripts/inspect_longport_token_expiry.py`
+- Pre-expiry refresh job (manual, dry-run default): `.github/workflows/refresh-longport-token.yml` + `scripts/refresh_longport_access_token.py`

scripts/refresh_longport_access_token.py

@@ -0,0 +1,271 @@
+#!/usr/bin/env python3
+"""Pre-expiry LongPort Legacy Access Token refresh (Actions / ops).
+
+Default is dry-run: report days remaining and whether a refresh would run.
+Live mode calls GET /v1/token/refresh with required ``expired_at``, then adds a
+Secret Manager version. Never prints secret values.
+
+Rollout order: paper → sg → hk last. Prefer dry-run until paper/sg succeed.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import hmac
+import json
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlencode
+
+_ROOT = Path(__file__).resolve().parents[1]
+if str(_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ROOT))
+
+from scripts.inspect_longport_token_expiry import (
+    TARGET_SECRETS,
+    days_until_expiry,
+    decode_token_expiry_unix,
+)
+
+DEFAULT_REFRESH_THRESHOLD_DAYS = 30
+DEFAULT_NEW_TOKEN_LIFETIME_DAYS = 90
+APP_KEY_SECRETS = {
+    "paper": "longport-app-key-paper",
+    "hk": "longport-app-key-hk",
+    "sg": "longport-app-key-sg",
+}
+APP_SECRET_SECRETS = {
+    "paper": "longport-app-secret-paper",
+    "hk": "longport-app-secret-hk",
+    "sg": "longport-app-secret-sg",
+}
+
+
+def _access_secret_latest(project_id: str, secret_name: str) -> str:
+    try:
+        import google.cloud.secretmanager_v1 as secret_manager
+    except ImportError:  # pragma: no cover
+        from google.cloud import secret_manager
+
+    client = secret_manager.SecretManagerServiceClient()
+    name = f"projects/{project_id}/secrets/{secret_name}/versions/latest"
+    response = client.access_secret_version(request={"name": name})
+    return response.payload.data.decode("UTF-8").strip()
+
+
+def _add_secret_version(project_id: str, secret_name: str, payload: str) -> str:
+    try:
+        import google.cloud.secretmanager_v1 as secret_manager
+    except ImportError:  # pragma: no cover
+        from google.cloud import secret_manager
+
+    client = secret_manager.SecretManagerServiceClient()
+    parent = f"projects/{project_id}/secrets/{secret_name}"
+    version = client.add_secret_version(
+        request={"parent": parent, "payload": {"data": payload.encode("UTF-8")}}
+    )
+    return str(version.name)
+
+
+def _format_refresh_expired_at(*, lifetime_days: int = DEFAULT_NEW_TOKEN_LIFETIME_DAYS) -> str:
+    expiry = datetime.now(timezone.utc) + timedelta(days=int(lifetime_days))
+    return expiry.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+
+def _longport_sign(method: str, uri: str, headers: dict[str, str], params: str, body: str, secret: str) -> str:
+    canonical_request = (
+        f"{method.upper()}|{uri}|{params}|"
+        f"authorization:{headers['Authorization']}\n"
+        f"x-api-key:{headers['X-Api-Key']}\n"
+        f"x-timestamp:{headers['X-Timestamp']}\n|authorization;x-api-key;x-timestamp|"
+    )
+    if body:
+        canonical_request += hashlib.sha1(body.encode("utf-8")).hexdigest()
+    sign_str = "HMAC-SHA256|" + hashlib.sha1(canonical_request.encode("utf-8")).hexdigest()
+    signature = hmac.new(secret.encode("utf-8"), sign_str.encode("utf-8"), hashlib.sha256).hexdigest()
+    return f"HMAC-SHA256 SignedHeaders=authorization;x-api-key;x-timestamp, Signature={signature}"
+
+
+def call_longport_refresh(
+    *,
+    access_token: str,
+    app_key: str,
+    app_secret: str,
+    lifetime_days: int = DEFAULT_NEW_TOKEN_LIFETIME_DAYS,
+    requests_module: Any | None = None,
+) -> str:
+    if requests_module is None:
+        import requests as requests_module
+
+    expired_at = _format_refresh_expired_at(lifetime_days=lifetime_days)
+    params = urlencode({"expired_at": expired_at})
+    headers = {
+        "X-Api-Key": app_key,
+        "Authorization": access_token,
+        "X-Timestamp": str(int(time.time())),
+        "Content-Type": "application/json; charset=utf-8",
+    }
+    headers["X-Api-Signature"] = _longport_sign("GET", "/v1/token/refresh", headers, params, "", app_secret)
+    response = requests_module.get(
+        f"https://openapi.longportapp.com/v1/token/refresh?{params}",
+        headers=headers,
+        timeout=15,
+    ).json()
+    if response.get("code") != 0:
+        code = response.get("code")
+        message = response.get("message") or "unknown error"
+        raise RuntimeError(f"LongPort refresh failed with code {code}: {message}")
+    data = response.get("data") or {}
+    new_token = data.get("token")
+    if not isinstance(new_token, str) or not new_token.strip():
+        raise RuntimeError("LongPort refresh returned an empty token payload")
+    return new_token.strip()
+
+
+def plan_refresh(
+    *,
+    days_remaining: float | None,
+    refresh_threshold_days: float,
+    force: bool,
+) -> tuple[bool, str]:
+    if force:
+        return True, "forced"
+    if days_remaining is None:
+        return True, "jwt_exp_undecodable"
+    if days_remaining <= 0:
+        return False, "already_expired_needs_portal_reset"
+    if days_remaining < float(refresh_threshold_days):
+        return True, "within_threshold"
+    return False, "outside_threshold"
+
+
+def run_refresh(
+    *,
+    project_id: str,
+    target: str,
+    dry_run: bool,
+    refresh_threshold_days: float = DEFAULT_REFRESH_THRESHOLD_DAYS,
+    force: bool = False,
+    lifetime_days: int = DEFAULT_NEW_TOKEN_LIFETIME_DAYS,
+    secret_reader: Any | None = None,
+    secret_writer: Any | None = None,
+    refresh_caller: Any | None = None,
+    now: float | None = None,
+) -> dict[str, Any]:
+    if target not in TARGET_SECRETS:
+        raise ValueError(f"unsupported target={target}")
+    token_secret = TARGET_SECRETS[target]
+    app_key_secret = APP_KEY_SECRETS[target]
+    app_secret_secret = APP_SECRET_SECRETS[target]
+    reader = secret_reader or _access_secret_latest
+    writer = secret_writer or _add_secret_version
+    refresher = refresh_caller or call_longport_refresh
+
+    current = now if now is not None else time.time()
+    token = reader(project_id, token_secret)
+    expiry = decode_token_expiry_unix(token)
+    days = None if expiry is None else days_until_expiry(expiry, now=current)
+    should, reason = plan_refresh(
+        days_remaining=days,
+        refresh_threshold_days=refresh_threshold_days,
+        force=force,
+    )
+    result: dict[str, Any] = {
+        "target": target,
+        "token_secret": token_secret,
+        "dry_run": bool(dry_run),
+        "force": bool(force),
+        "refresh_threshold_days": float(refresh_threshold_days),
+        "jwt_exp_present": expiry is not None,
+        "days_until_expiry": None if days is None else round(days, 3),
+        "would_refresh": bool(should),
+        "reason": reason,
+        "status": "planned",
+        "ok": True,
+    }
+    if not should:
+        if reason == "already_expired_needs_portal_reset":
+            result["status"] = "blocked_expired"
+            result["ok"] = False
+        else:
+            result["status"] = "skipped"
+        return result
+
+    if dry_run:
+        result["status"] = "dry_run_would_refresh"
+        return result
+
+    app_key = reader(project_id, app_key_secret)
+    app_secret = reader(project_id, app_secret_secret)
+    if not app_key or not app_secret:
+        result["status"] = "missing_app_credentials"
+        result["ok"] = False
+        return result
+
+    new_token = refresher(
+        access_token=token,
+        app_key=app_key,
+        app_secret=app_secret,
+        lifetime_days=lifetime_days,
+    )
+    version_name = writer(project_id, token_secret, new_token)
+    result["status"] = "refreshed"
+    result["new_version"] = version_name
+    return result
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--project", required=True)
+    parser.add_argument("--target", required=True, choices=sorted(TARGET_SECRETS))
+    parser.add_argument(
+        "--dry-run",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help="Default true: plan only; never call LongPort refresh or write SM",
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Refresh even when days remaining >= threshold (still blocked if already expired)",
+    )
+    parser.add_argument(
+        "--refresh-threshold-days",
+        type=float,
+        default=DEFAULT_REFRESH_THRESHOLD_DAYS,
+    )
+    parser.add_argument(
+        "--new-token-lifetime-days",
+        type=int,
+        default=DEFAULT_NEW_TOKEN_LIFETIME_DAYS,
+    )
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    if args.target == "hk" and not args.dry_run:
+        print(
+            "warning: live refresh on target=hk invalidates the current Access Token; "
+            "prefer paper then sg first",
+            file=sys.stderr,
+        )
+    result = run_refresh(
+        project_id=args.project,
+        target=args.target,
+        dry_run=args.dry_run,
+        refresh_threshold_days=args.refresh_threshold_days,
+        force=args.force,
+        lifetime_days=args.new_token_lifetime_days,
+    )
+    print(json.dumps(result, sort_keys=True))
+    return 0 if result.get("ok") else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())