google-workspace

1#!/usr/bin/env python32"""Bridge between Hermes OAuth token and gws CLI.3 4Refreshes the token if expired, then executes gws with the valid access token.5"""6import json7import os8import subprocess9import sys10from datetime import datetime, timezone11from pathlib import Path12 13# Ensure sibling modules (_hermes_home) are importable when run standalone.14_SCRIPTS_DIR = str(Path(__file__).resolve().parent)15if _SCRIPTS_DIR not in sys.path:16    sys.path.insert(0, _SCRIPTS_DIR)17 18from _hermes_home import get_hermes_home19 20 21def get_token_path() -> Path:22    return get_hermes_home() / "google_token.json"23 24 25def _normalize_authorized_user_payload(payload: dict) -> dict:26    normalized = dict(payload)27    if not normalized.get("type"):28        normalized["type"] = "authorized_user"29    return normalized30 31 32def refresh_token(token_data: dict) -> dict:33    """Refresh the access token using the refresh token."""34    import urllib.error35    import urllib.parse36    import urllib.request37 38    required_keys = ["client_id", "client_secret", "refresh_token", "token_uri"]39    missing = [k for k in required_keys if k not in token_data]40    if missing:41        print(f"ERROR: google_token.json is missing required fields: {', '.join(missing)}", file=sys.stderr)42        print("Please re-authenticate by running the Google Workspace setup script.", file=sys.stderr)43        sys.exit(1)44 45    params = urllib.parse.urlencode({46        "client_id": token_data["client_id"],47        "client_secret": token_data["client_secret"],48        "refresh_token": token_data["refresh_token"],49        # The refresh token goes in BOTH the body and the ``x-nous-refresh-token`` header. Portal's token50        # endpoint requires ``refresh_token`` in the body (its request schema rejects a header-only request51        # as ``invalid_request``), and additionally reconciles the header against the body — sending both52        # lets Portal keep the value out of body-access-logs while still satisfying the schema. The header53        # name must match Portal's ``REFRESH_TOKEN_HEADER`` exactly (``x-nous-refresh- token``); any other54        # name is silently ignored. (Verified against the NAS #293 preview deploy: header-only → 40055        # invalid_request; body → accepted.)56        "grant_type": "refresh_token",57    }).encode()58 59    req = urllib.request.Request(token_data["token_uri"], data=params)60    try:61        with urllib.request.urlopen(req, timeout=15) as resp:62            result = json.loads(resp.read())63    except urllib.error.HTTPError as e:64        body = e.read().decode("utf-8", errors="replace")65        print(f"ERROR: Token refresh failed (HTTP {e.code}): {body}", file=sys.stderr)66        print("Re-run setup.py to re-authenticate.", file=sys.stderr)67        sys.exit(1)68    except (urllib.error.URLError, TimeoutError) as e:69        print(f"ERROR: Token refresh failed (network): {e}", file=sys.stderr)70        sys.exit(1)71 72    token_data["token"] = result["access_token"]73    token_data["expiry"] = datetime.fromtimestamp(74        datetime.now(timezone.utc).timestamp() + result["expires_in"],75        tz=timezone.utc,76    ).isoformat()77 78    get_token_path().write_text(79        json.dumps(_normalize_authorized_user_payload(token_data), indent=2), encoding="utf-8"80    )81    return token_data82 83 84def get_valid_token() -> str:85    """Return a valid access token, refreshing if needed."""86    token_path = get_token_path()87    if not token_path.exists():88        print("ERROR: No Google token found. Run setup.py --auth-url first.", file=sys.stderr)89        sys.exit(1)90 91    token_data = json.loads(token_path.read_text(encoding="utf-8"))92 93    expiry = token_data.get("expiry", "")94    if expiry:95        exp_dt = datetime.fromisoformat(expiry.replace("Z", "+00:00"))96        now = datetime.now(timezone.utc)97        if now >= exp_dt:98            token_data = refresh_token(token_data)99 100    return token_data["token"]101 102 103def main():104    """Refresh token if needed, then exec gws with remaining args."""105    if len(sys.argv) < 2:106        print("Usage: gws_bridge.py <gws args...>", file=sys.stderr)107        sys.exit(1)108 109    access_token = get_valid_token()110    env = os.environ.copy()111    env["GOOGLE_WORKSPACE_CLI_TOKEN"] = access_token112 113    result = subprocess.run(["gws"] + sys.argv[1:], env=env)114    sys.exit(result.returncode)115 116 117if __name__ == "__main__":118    main()119