Spaces:
Runtime error
Runtime error
File size: 2,338 Bytes
6d34792 d584009 6d34792 40776c4 6d34792 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """Shared Google OAuth auth-gate helpers for Drive and Calendar tools."""
from google.auth.exceptions import RefreshError
from googleapiclient.errors import HttpError
from google_auth_flow import credentials_from_token_dict, get_auth_url
from token_store import delete_token, get_token, save_token
AUTH_REQUIRED_PREFIX = "AUTH_REQUIRED::"
def is_auth_failure(exc: BaseException) -> bool:
if isinstance(exc, RefreshError):
return True
if "invalid_grant" in str(exc).lower():
return True
if isinstance(exc, HttpError):
status = exc.resp.status if exc.resp else None
if status in (401, 403):
return True
if "invalid_grant" in str(exc).lower():
return True
return False
def auth_required_message(
user_email: str,
product: str = "Google",
*,
revoke: bool = False,
) -> str:
if revoke:
try:
delete_token(user_email)
except Exception:
pass
auth_url = get_auth_url(user_email)
return (
f"{AUTH_REQUIRED_PREFIX}{auth_url}\n"
f"User {user_email} must sign in to grant {product} access. "
f"Visit the URL above, then retry the request."
)
def get_google_credentials(user_email: str):
"""
Load and refresh stored credentials.
Returns (credentials, None) on success, (None, None) if no token,
or (None, auth_required_message) when re-authentication is needed.
"""
token_dict = get_token(user_email)
if not token_dict:
return None, None
try:
creds = credentials_from_token_dict(token_dict)
save_token(
user_email,
{
"token": creds.token,
"refresh_token": (
creds.refresh_token
or token_dict.get("refresh_token")
),
"token_uri": creds.token_uri,
"client_id": creds.client_id,
"client_secret": creds.client_secret,
"scopes": list(creds.scopes or []),
"expiry": creds.expiry.isoformat() if creds.expiry else None,
},
)
return creds, None
except Exception as e:
if is_auth_failure(e):
return None, auth_required_message(user_email, "Google", revoke=True)
raise
|