gitea-admin/gh-christianlouis-inboxconverge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: proactive Gmail token refresh, GmailAuthError, improved logging

Browse Source 
Agent-Logs-Url: https://github.com/christianlouis/InboxConverge/sessions/2bcd611e-3598-4b83-95bd-3276a7d5ad7c

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-05-03 14:50:58 +00:00
committed by GitHub
parent d00523f5ba
commit 992e157f02
4 changed files with 360 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/gmail_service.py
+113 -1
View File
@@ -11,11 +11,13 @@ import base64
import logging
import textwrap
import time
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from email.mime.text import MIMEText
from email.utils import format_datetime
from typing import Optional, Dict, Any
import google.auth.exceptions
from google.auth.transport.requests import Request as GoogleAuthRequest
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
@@ -43,6 +45,12 @@ class GmailInjectionError(Exception):
pass
class GmailAuthError(GmailInjectionError):
"""Raised when Gmail OAuth2 authentication fails (token expired or revoked)."""
pass
class GmailService:
"""
Service for injecting emails into Gmail via the Gmail API.
@@ -161,6 +169,16 @@ class GmailService:
logger.error(error_msg)
# Surface 401 so callers can mark credentials as invalid
raise GmailInjectionError(error_msg)
except google.auth.exceptions.RefreshError as e:
_dur = time.perf_counter() - _start
GMAIL_API_REQUESTS_TOTAL.labels(operation="inject", status="error").inc()
GMAIL_API_DURATION_SECONDS.labels(operation="inject").observe(_dur)
error_msg = (
f"Gmail token refresh failed — the refresh token may have been revoked. "
f"The user must re-authorise. Detail: {e}"
)
logger.error(error_msg)
raise GmailAuthError(error_msg)
except Exception as e:
_dur = time.perf_counter() - _start
GMAIL_API_REQUESTS_TOTAL.labels(operation="inject", status="error").inc()
@@ -190,6 +208,15 @@ class GmailService:
email = result.get("emailAddress", "unknown")
logger.info(f"Gmail API access verified for: {email}")
return True
except google.auth.exceptions.RefreshError as e:
_dur = time.perf_counter() - _start
GMAIL_API_REQUESTS_TOTAL.labels(operation="verify", status="error").inc()
GMAIL_API_DURATION_SECONDS.labels(operation="verify").observe(_dur)
logger.error(
f"Gmail token refresh failed during access verification — "
f"refresh token may be revoked. Detail: {e}"
)
return False
except Exception as e:
_dur = time.perf_counter() - _start
GMAIL_API_REQUESTS_TOTAL.labels(operation="verify", status="error").inc()
@@ -286,6 +313,16 @@ class GmailService:
error_msg = f"Gmail API error while managing label '{name}': {e.reason if hasattr(e, 'reason') else str(e)}"
logger.error(error_msg)
raise GmailInjectionError(error_msg)
except google.auth.exceptions.RefreshError as e:
_dur = time.perf_counter() - _start
GMAIL_API_REQUESTS_TOTAL.labels(operation="get_label", status="error").inc()
GMAIL_API_DURATION_SECONDS.labels(operation="get_label").observe(_dur)
error_msg = (
f"Gmail token refresh failed while managing label '{name}' — "
f"refresh token may be revoked. Detail: {e}"
)
logger.error(error_msg)
raise GmailAuthError(error_msg)
except Exception as e:
_dur = time.perf_counter() - _start
GMAIL_API_REQUESTS_TOTAL.labels(operation="get_label", status="error").inc()
@@ -376,6 +413,81 @@ class GmailService:
return label_ids
def is_token_expiring_soon(self, within_minutes: int = 30) -> bool:
"""
Return True if the access token has already expired or will expire
within *within_minutes* minutes.
When ``credentials.expiry`` is None the expiry is unknown, which is
treated as expiring soon so a proactive refresh is performed.
Args:
within_minutes: Threshold in minutes before which a token is
considered "expiring soon".
Returns:
True if the token needs refreshing, False otherwise.
"""
if self.credentials.expiry is None:
return True
threshold = datetime.now(timezone.utc) + timedelta(minutes=within_minutes)
expiry = self.credentials.expiry
# google-auth stores expiry as a naive UTC datetime; make it tz-aware.
if expiry.tzinfo is None:
expiry = expiry.replace(tzinfo=timezone.utc)
return expiry <= threshold
async def proactive_refresh(self) -> Dict[str, Any]:
"""
Explicitly refresh the access token using the stored refresh token.
Unlike the lazy refresh that happens automatically during API calls,
this method triggers a refresh regardless of whether the current
access token has expired. Use this from a scheduled task to keep
tokens fresh and to detect revocation early.
Returns:
Dict with ``access_token`` and ``expiry`` (datetime | None)
representing the newly obtained access token.
Raises:
GmailAuthError: If the refresh token is missing, has been revoked,
or the refresh request fails for an auth-related reason.
GmailInjectionError: For unexpected non-auth errors.
"""
if not self.credentials.refresh_token:
raise GmailAuthError(
"Cannot refresh: no refresh token stored. "
"Re-authorise Gmail to obtain a new refresh token."
)
loop = asyncio.get_event_loop()
try:
await loop.run_in_executor(
None,
lambda: self.credentials.refresh(GoogleAuthRequest()),
)
GMAIL_TOKEN_REFRESHES_TOTAL.inc()
logger.info(
"Gmail access token refreshed proactively; " "new expiry: %s",
self.credentials.expiry,
)
return {
"access_token": self.credentials.token,
"expiry": self.credentials.expiry,
}
except google.auth.exceptions.RefreshError as e:
error_msg = (
f"Gmail refresh token has been revoked or is invalid — "
f"the user must re-authorise. Detail: {e}"
)
logger.error(error_msg)
raise GmailAuthError(error_msg)
except Exception as e:
raise GmailInjectionError(
f"Unexpected error during Gmail token refresh: {e}"
)
def get_refreshed_token(self) -> Optional[Dict[str, Any]]:
"""
Return the current access token and expiry if the token was refreshed
backend/app/workers/celery_app.py
+4
View File
@@ -43,6 +43,10 @@ celery_app.conf.beat_schedule = {
minute="*"
), # Every minute (per-account interval gates actual work)
},
"refresh-gmail-tokens": {
"task": "app.workers.tasks.refresh_gmail_tokens",
"schedule": crontab(minute="*/45"), # Every 45 minutes
},
"cleanup-old-logs": {
"task": "app.workers.tasks.cleanup_old_logs",
"schedule": crontab(hour=3, minute=0), # Daily at 3 AM
backend/app/workers/tasks.py
+236 -8
View File
@@ -32,11 +32,11 @@ from app.models.database_models import (
UserSmtpConfig,
)
from app.services.mail_processor import MailProcessor
from app.services.gmail_service import GmailService
from app.services.gmail_service import GmailService, GmailAuthError
from app.services.config_service import ConfigService
from app.services.notification_service import send_user_notification
from app.core.config import settings
from sqlalchemy import select, delete
from sqlalchemy import select, delete, or_
logger = logging.getLogger(__name__)
@@ -156,6 +156,27 @@ async def process_mail_account(account_id: int):
if gmail_cred.encrypted_refresh_token
else None
)
# Log token expiry to help diagnose timeout issues.
if gmail_cred.token_expiry:
_now = datetime.now(timezone.utc)
_expiry = gmail_cred.token_expiry
if _expiry.tzinfo is None:
_expiry = _expiry.replace(tzinfo=timezone.utc)
_secs = (_expiry - _now).total_seconds()
if _secs < 0:
logger.info(
"Gmail access token for user %s expired %.0f s ago; "
"google-auth will refresh automatically using the refresh token.",
account.user_id,
-_secs,
)
elif _secs < 300:
logger.info(
"Gmail access token for user %s expires in %.0f s; "
"will be refreshed on the first API call.",
account.user_id,
_secs,
)
gmail_service = GmailService(
access_token=access_token,
refresh_token=refresh_token,
@@ -324,10 +345,46 @@ async def process_mail_account(account_id: int):
)
emails_failed += 1
except GmailAuthError as e:
# Token refresh failed: refresh token is likely revoked.
# Mark the credential invalid so the user is prompted to
# re-authorise, and stop retrying for this run.
if use_gmail_api and gmail_cred:
gmail_cred.is_valid = False # type: ignore[assignment]
GMAIL_CREDENTIALS_INVALIDATED_TOTAL.inc()
logger.warning(
"Gmail credentials revoked for user %s "
"(account %s, token_expiry=%s). "
"User must re-authorise. Error: %s",
account.user_id,
account.id,
getattr(gmail_cred, "token_expiry", "unknown"),
e,
)
try:
async with async_session_maker() as notif_db:
await send_user_notification(
db=notif_db,
user_id=int(account.user_id),
title="InboxRescue: Gmail Authorization Expired",
body=(
f"Your Gmail credentials for account '{account.name}' "
f"have been revoked or have expired. "
f"Please re-authorize Gmail access in Settings."
),
notify_on_error=True,
)
except Exception as notify_exc:
logger.warning(
"Failed to send revocation notification: %s", notify_exc
)
error_msg = str(e)
emails_failed += 1
except Exception as e:
error_str = str(e).lower()
# If Gmail returns 401/403 the refresh token was revoked –
# mark credentials invalid so the user gets notified.
# Catch any remaining auth-style errors that slipped through
# (e.g. HttpError 401/403 returned after a successful refresh).
if (
use_gmail_api
and gmail_cred
@@ -340,8 +397,13 @@ async def process_mail_account(account_id: int):
gmail_cred.is_valid = False # type: ignore[assignment]
GMAIL_CREDENTIALS_INVALIDATED_TOTAL.inc()
logger.warning(
f"Gmail credentials revoked for user {account.user_id}. "
"User must re-authorise."
"Gmail credentials invalidated for user %s "
"(account %s, token_expiry=%s) due to HTTP auth error. "
"User must re-authorise. Error: %s",
account.user_id,
account.id,
getattr(gmail_cred, "token_expiry", "unknown"),
e,
)
try:
async with async_session_maker() as notif_db:
@@ -349,14 +411,23 @@ async def process_mail_account(account_id: int):
db=notif_db,
user_id=int(account.user_id),
title="InboxRescue: Gmail Authorization Expired",
body=f"Your Gmail credentials for account '{account.name}' have been revoked. Please re-authorize Gmail access in Settings.",
body=(
f"Your Gmail credentials for account '{account.name}' "
f"have been revoked. Please re-authorize Gmail access "
f"in Settings."
),
notify_on_error=True,
)
except Exception as notify_exc:
logger.warning(
f"Failed to send revocation notification: {notify_exc}"
)
logger.error(f"Error delivering email: {e}")
logger.error(
"Error delivering email (account %s, uid=%s): %s",
account.id,
uid,
e,
)
error_msg = str(e)
emails_failed += 1
@@ -728,3 +799,160 @@ async def cleanup_old_logs(days_to_keep: int = 30):
CELERY_TASK_DURATION_SECONDS.labels(task_name="cleanup_old_logs").observe(
_task_duration
)
@celery_app.task(base=AsyncTask, name="app.workers.tasks.refresh_gmail_tokens")
async def refresh_gmail_tokens():
"""
Proactively refresh Gmail OAuth2 access tokens that are close to expiry.
Runs every 45 minutes via Celery Beat. Any credential whose access token
expires within the next 30 minutes (or whose expiry is unknown) is
refreshed using its stored refresh token.
Benefits:
- Keeps the DB's ``encrypted_access_token`` and ``token_expiry`` columns
up to date so mail-processing tasks never start with an already-expired
token (which would cause a delayed in-band refresh).
- Detects revoked refresh tokens early, before they block email delivery,
and sends the user a re-authorisation notification.
Credentials without a refresh token are skipped — they cannot be refreshed
automatically and will simply fail on the next delivery attempt.
"""
_task_start = time.monotonic()
async with async_session_maker() as db:
try:
# Find all valid credentials with a refresh token whose access
# token expires within the next 30 minutes (or expiry is unknown).
refresh_threshold = datetime.now(timezone.utc) + timedelta(minutes=30)
cred_result = await db.execute(
select(GmailCredential).where(
GmailCredential.is_valid == True, # noqa: E712
GmailCredential.encrypted_refresh_token.is_not(None),
or_(
GmailCredential.token_expiry.is_(None),
GmailCredential.token_expiry <= refresh_threshold,
),
)
)
credentials_to_refresh = cred_result.scalars().all()
if not credentials_to_refresh:
logger.debug("refresh_gmail_tokens: no credentials need refreshing")
_task_duration = time.monotonic() - _task_start
CELERY_TASKS_TOTAL.labels(
task_name="refresh_gmail_tokens", status="success"
).inc()
CELERY_TASK_DURATION_SECONDS.labels(
task_name="refresh_gmail_tokens"
).observe(_task_duration)
return
logger.info(
"refresh_gmail_tokens: refreshing %d credential(s)",
len(credentials_to_refresh),
)
refreshed_count = 0
failed_count = 0
for cred in credentials_to_refresh:
access_token = decrypt_credential(cred.encrypted_access_token) # type: ignore[arg-type]
refresh_token = decrypt_credential(cred.encrypted_refresh_token) # type: ignore[arg-type]
gmail_service = GmailService(
access_token=access_token,
refresh_token=refresh_token,
client_id=settings.GOOGLE_CLIENT_ID,
client_secret=settings.GOOGLE_CLIENT_SECRET,
)
try:
new_token_info = await gmail_service.proactive_refresh()
cred.encrypted_access_token = encrypt_credential( # type: ignore[assignment]
new_token_info["access_token"]
)
if new_token_info.get("expiry"):
cred.token_expiry = new_token_info["expiry"] # type: ignore[assignment]
cred.last_verified_at = datetime.now(timezone.utc) # type: ignore[assignment]
logger.info(
"refresh_gmail_tokens: refreshed token for user %s "
"(gmail=%s, new_expiry=%s)",
cred.user_id,
cred.gmail_email,
new_token_info.get("expiry"),
)
refreshed_count += 1
except GmailAuthError as e:
# Refresh token is revoked — mark the credential invalid
# and notify the user.
cred.is_valid = False # type: ignore[assignment]
GMAIL_CREDENTIALS_INVALIDATED_TOTAL.inc()
logger.warning(
"refresh_gmail_tokens: refresh token revoked for user %s "
"(gmail=%s). Marking invalid. Error: %s",
cred.user_id,
cred.gmail_email,
e,
)
failed_count += 1
try:
async with async_session_maker() as notif_db:
await send_user_notification(
db=notif_db,
user_id=int(cred.user_id),
title="InboxConverge: Gmail Re-authorisation Required",
body=(
"Your Gmail access has been revoked. "
"Please open Settings → Gmail API and click "
"'Connect Gmail' to restore email delivery."
),
notify_on_error=True,
)
except Exception as notify_exc:
logger.warning(
"refresh_gmail_tokens: failed to send revocation "
"notification for user %s: %s",
cred.user_id,
notify_exc,
)
except Exception as e:
# Non-auth error (e.g. network timeout) — log but do not
# mark as invalid; it may succeed on the next run.
logger.warning(
"refresh_gmail_tokens: unexpected error refreshing token "
"for user %s (gmail=%s): %s",
cred.user_id,
cred.gmail_email,
e,
)
failed_count += 1
await db.commit()
logger.info(
"refresh_gmail_tokens: finished — %d refreshed, %d failed",
refreshed_count,
failed_count,
)
_task_duration = time.monotonic() - _task_start
CELERY_TASKS_TOTAL.labels(
task_name="refresh_gmail_tokens", status="success"
).inc()
CELERY_TASK_DURATION_SECONDS.labels(
task_name="refresh_gmail_tokens"
).observe(_task_duration)
except Exception as e:
logger.error("refresh_gmail_tokens: unexpected error: %s", e)
_task_duration = time.monotonic() - _task_start
CELERY_TASKS_TOTAL.labels(
task_name="refresh_gmail_tokens", status="failure"
).inc()
CELERY_TASK_DURATION_SECONDS.labels(
task_name="refresh_gmail_tokens"
).observe(_task_duration)