This commit is contained in:
levxn
2026-03-20 22:32:30 +05:30
parent 1f12a45151
commit 07f7801166
2 changed files with 53 additions and 15 deletions
+39 -1
View File
@@ -306,6 +306,38 @@ def get_antigravity_client_id() -> str:
return "1071006060591-tmhssin2h21lcre235vtolojh4g403ep.apps.googleusercontent.com"
def _read_antigravity_secret_from_npm() -> str | None:
"""Read the Antigravity OAuth client secret from the globally installed npm package.
Mirrors the bash ``_read_antigravity_creds_from_npm()`` helper so that users
who have ``opencode-antigravity-auth`` installed globally always get the secret
at runtime regardless of whether quickstart wrote it to their config.
"""
import re as _re
import subprocess
from pathlib import Path as _Path
candidates: list[_Path] = []
try:
npm_root = subprocess.check_output(
["npm", "root", "-g"], stderr=subprocess.DEVNULL, timeout=5
).decode().strip()
candidates.append(_Path(npm_root) / "opencode-antigravity-auth/dist/src/constants.js")
except Exception:
pass
candidates += [
_Path("/opt/homebrew/lib/node_modules/opencode-antigravity-auth/dist/src/constants.js"),
_Path("/usr/local/lib/node_modules/opencode-antigravity-auth/dist/src/constants.js"),
_Path("/usr/lib/node_modules/opencode-antigravity-auth/dist/src/constants.js"),
]
for p in candidates:
if p.exists():
m = _re.search(r'"(GOCSPX-[^"]+)"', p.read_text(errors="ignore"))
if m:
return m.group(1)
return None
def get_antigravity_client_secret() -> str | None:
"""Return the Antigravity OAuth client secret.
@@ -313,6 +345,7 @@ def get_antigravity_client_secret() -> str | None:
1. ``ANTIGRAVITY_CLIENT_SECRET`` environment variable
2. ``llm.antigravity_client_secret`` in ~/.hive/configuration.json
(written by quickstart when Antigravity is configured)
3. Globally installed ``opencode-antigravity-auth`` npm package (runtime fallback)
Returns None when not found token refresh will be skipped and
the caller must use whatever access token is already available.
@@ -320,7 +353,12 @@ def get_antigravity_client_secret() -> str | None:
env = os.environ.get("ANTIGRAVITY_CLIENT_SECRET")
if env:
return env
return get_hive_config().get("llm", {}).get("antigravity_client_secret") or None
cfg_val = get_hive_config().get("llm", {}).get("antigravity_client_secret") or None
if cfg_val:
return cfg_val
# Runtime fallback: read from globally installed npm package so users who set up
# via a path other than the updated quickstart still get the secret automatically.
return _read_antigravity_secret_from_npm()
def get_gcu_enabled() -> bool:
+14 -14
View File
@@ -200,21 +200,21 @@ def _load_from_ide_db() -> tuple[str | None, str | None, float]:
def _do_token_refresh(refresh_token: str) -> tuple[str, float] | None:
"""POST to Google OAuth endpoint and return ``(new_access_token, expires_at)``.
Requires the Antigravity OAuth client secret to be configured via:
- ``ANTIGRAVITY_CLIENT_SECRET`` environment variable, or
- ``llm.antigravity_client_secret`` in ~/.hive/configuration.json
The client secret is sourced via ``get_antigravity_client_secret()`` (env var,
config file, or npm package fallback). When unavailable the refresh is attempted
without it Google will reject it for web-app clients, but the npm fallback in
``get_antigravity_client_secret()`` should ensure the secret is found at runtime.
Returns None (skips refresh) when the secret is unavailable.
Returns None when the HTTP request fails.
"""
from framework.config import get_antigravity_client_secret # noqa: PLC0415
client_secret = get_antigravity_client_secret()
if not client_secret:
logger.debug(
"Antigravity client secret not configured — skipping token refresh. "
"Antigravity client secret not configured — attempting refresh without it. "
"Set ANTIGRAVITY_CLIENT_SECRET or run quickstart to configure."
)
return None
import urllib.error # noqa: PLC0415
import urllib.parse # noqa: PLC0415
@@ -222,14 +222,14 @@ def _do_token_refresh(refresh_token: str) -> tuple[str, float] | None:
from framework.config import get_antigravity_client_id # noqa: PLC0415
body = urllib.parse.urlencode(
{
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": get_antigravity_client_id(),
"client_secret": client_secret,
}
).encode("utf-8")
params: dict[str, str] = {
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": get_antigravity_client_id(),
}
if client_secret:
params["client_secret"] = client_secret
body = urllib.parse.urlencode(params).encode("utf-8")
req = urllib.request.Request(
_TOKEN_URL,