* [PATCH kgl 1/2] Decouple GmailTarget from OAuth file credential management
2026-11-11 6:59 [PATCH kgl 0/2] Add token_command support for Gmail targets Dmitry Torokhov
@ 2026-11-11 6:59 ` Dmitry Torokhov
2026-11-11 6:59 ` [PATCH kgl 2/2] Add token_command support for Gmail targets Dmitry Torokhov
2026-11-12 19:58 ` [PATCH kgl 0/2] " Konstantin Ryabitsev
2 siblings, 0 replies; 4+ messages in thread
From: Dmitry Torokhov @ 2026-11-11 6:59 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev
Move OAuth client secrets and token file management out of GmailTarget
into OAuthFileCredentials in google_creds, subclassing
google.auth.credentials.Credentials, and pass credentials directly into
GmailTarget via make_gmail_credentials.
This keeps Google OAuth credential transport separate from Gmail API
delivery and configuration parsing, and prepares for supporting
alternative credential providers.
Signed-off-by: Dmitry Torokhov <dmitry.torokhov@gmail.com>
---
src/korgalore/cli.py | 23 ++---
src/korgalore/gmail_target.py | 212 ++++++++++++++++++--------------------
src/korgalore/google_creds.py | 229 ++++++++++++++++++++++++++++++++++++++++++
tests/test_gmail_target.py | 229 +++++++++++++++++++++---------------------
tests/test_google_creds.py | 154 ++++++++++++++++++++++++++++
5 files changed, 610 insertions(+), 237 deletions(-)
diff --git a/src/korgalore/cli.py b/src/korgalore/cli.py
index 3e3e615..180906d 100644
--- a/src/korgalore/cli.py
+++ b/src/korgalore/cli.py
@@ -54,7 +54,7 @@ from korgalore.digest import (
strip_reply_prefixes,
)
from korgalore.dummy_target import DummyTarget
-from korgalore.gmail_target import GmailTarget
+from korgalore.gmail_target import GmailTarget, make_gmail_credentials
from korgalore.imap_target import ImapTarget
from korgalore.jmap_target import JmapTarget
from korgalore.lei_feed import LeiFeed
@@ -280,8 +280,7 @@ def get_target(ctx: click.Context, identifier: str) -> Any:
if target_type == 'gmail':
service = get_gmail_target(
identifier=identifier,
- credentials_file=details.get('credentials', ''),
- token_file=details.get('token', None),
+ details=details,
interactive=interactive,
)
elif target_type == 'maildir':
@@ -334,19 +333,19 @@ def get_target(ctx: click.Context, identifier: str) -> Any:
def get_gmail_target(
- identifier: str, credentials_file: str, token_file: str | None, interactive: bool = True
+ identifier: str,
+ details: Mapping[str, Any],
+ interactive: bool = True,
) -> GmailTarget:
"""Create a Gmail target service instance."""
- if not credentials_file:
- logger.critical('No credentials file specified for Gmail target: %s', identifier)
- raise click.Abort()
- if not token_file:
- cfgdir = get_xdg_config_dir()
- token_file = str(cfgdir / f'gmail-{identifier}-token.json')
try:
- gt = GmailTarget(
- identifier=identifier, credentials_file=credentials_file, token_file=token_file, interactive=interactive
+ creds = make_gmail_credentials(
+ identifier=identifier,
+ details=details,
+ interactive=interactive,
+ config_dir=get_xdg_config_dir(),
)
+ gt = GmailTarget(identifier=identifier, creds=creds)
except ConfigurationError as fe:
logger.critical('Error: %s', str(fe))
raise click.Abort() from fe
diff --git a/src/korgalore/gmail_target.py b/src/korgalore/gmail_target.py
index c88c3d9..3fd9f8b 100644
--- a/src/korgalore/gmail_target.py
+++ b/src/korgalore/gmail_target.py
@@ -1,22 +1,21 @@
+import base64
import logging
import os
+from collections.abc import Mapping
from pathlib import Path
from typing import Any
-from google.auth.exceptions import RefreshError
-from google.auth.external_account_authorized_user import Credentials as ExternalAccountCredentials
+from google.auth.credentials import Credentials as BaseCredentials
from google.auth.transport.requests import Request
-from google.oauth2.credentials import Credentials
-from google_auth_oauthlib.flow import InstalledAppFlow # type: ignore
-from googleapiclient.discovery import build # type: ignore
-from googleapiclient.errors import HttpError # type: ignore
+from googleapiclient.discovery import Resource, build # type: ignore[import-untyped]
+from googleapiclient.errors import HttpError # type: ignore[import-untyped]
from korgalore import AuthenticationError, ConfigurationError, RemoteError
+from korgalore.google_creds import OAuthFileCredentials
from korgalore.message import RawMessage
logger = logging.getLogger('korgalore')
-
# If modifying these scopes, delete the file token.json.
# We need scopes for reading and inserting new emails, but not
# modifying existing ones.
@@ -25,105 +24,52 @@ SCOPES = [
'https://www.googleapis.com/auth/gmail.insert',
]
-# InstalledAppFlow.run_local_server() returns either flavour depending on what
-# the client_secrets file describes. Both expose the valid/expired/refresh_token
-# attributes and the to_json() and refresh() methods used below, so the code
-# does not care which one it gets -- but the declared type has to allow both.
-GmailCredentials = Credentials | ExternalAccountCredentials
-
class GmailTarget:
"""Target class for delivering email messages to Gmail via the API."""
DEFAULT_LABELS: list[str] = ['INBOX', 'UNREAD']
- def __init__(self, identifier: str, credentials_file: str, token_file: str, interactive: bool = True) -> None:
+ def __init__(self, identifier: str, creds: BaseCredentials) -> None:
"""Initialize a GmailTarget instance.
Args:
identifier: Unique identifier for this Gmail target.
- credentials_file: Path to the Google OAuth credentials JSON file.
- token_file: Path to store/load the OAuth token.
- interactive: If True, run OAuth flow interactively when needed.
- If False, raise AuthenticationError instead (for GUI mode).
-
- Raises:
- ConfigurationError: If credentials file is not found.
- AuthenticationError: If token is expired/revoked and re-auth is needed,
- or if interactive=False and no token exists.
+ creds: Google API credentials instance.
"""
self.identifier = identifier
- self.creds: GmailCredentials | None = None
- self.service: Any | None = None
+ self.creds = creds
+ self.service: Resource | None = None
self._label_map: dict[str, str] | None = None
- # Store expanded paths for potential re-authentication
- self._credentials_file = Path(os.path.expandvars(credentials_file)).expanduser()
- self._token_file = Path(os.path.expandvars(token_file)).expanduser()
- self._needs_auth = False
- self._interactive = interactive
- self._load_credentials()
-
- def _load_credentials(self) -> None:
- """Load or refresh OAuth credentials for Gmail API access.
- Attempts to load existing credentials from token_file. If not present
- or expired, initiates OAuth flow using credentials_file.
-
- Raises:
- ConfigurationError: If credentials file is not found.
- AuthenticationError: If token is expired/revoked and re-auth is needed.
- """
- # The file token.json stores the user's access and refresh tokens
- if self._token_file.exists():
- self.creds = Credentials.from_authorized_user_file(str(self._token_file), SCOPES) # type: ignore
-
- # If there are no (valid) credentials available, let the user log in
- if not self.creds or not self.creds.valid:
- if self.creds and self.creds.expired and self.creds.refresh_token:
- try:
- self.creds.refresh(Request()) # type: ignore
- except RefreshError as e:
- logger.warning('Gmail token for %s has expired or been revoked.', self.identifier)
- invalid_token_file = self._token_file.with_name(f'{self._token_file.name}.invalid')
- invalid_token_file.unlink(missing_ok=True)
- self._token_file.rename(invalid_token_file)
- self._needs_auth = True
- if not self._interactive:
- # In non-interactive mode, just return - caller will check needs_auth
- return
- raise AuthenticationError(
- f"Gmail token for '{self.identifier}' is invalid. Please re-authenticate.",
- target_id=self.identifier,
- target_type='gmail',
- ) from e
- elif self._credentials_file.exists():
- if not self._interactive:
- # In non-interactive mode (GUI), don't run OAuth flow
- # Just mark as needing auth and return - caller will check needs_auth
- self._needs_auth = True
- return
- logger.critical('Log in to Gmail account for %s', self.identifier)
-
- flow = InstalledAppFlow.from_client_secrets_file(str(self._credentials_file), SCOPES)
- self.creds = flow.run_local_server(port=0)
- else:
- raise ConfigurationError(
- f'{self._credentials_file} not found. Please download it from Google Cloud Console.'
- )
-
- # Save the credentials for the next run
- self._token_file.write_text(self.creds.to_json())
-
- self._needs_auth = False
+ @property
+ def needs_auth(self) -> bool:
+ """Check if this target needs re-authentication."""
+ return bool(getattr(self.creds, 'needs_auth', False))
def connect(self) -> None:
"""Establish connection to the Gmail API service.
Creates the Gmail API service object if not already connected.
+
+ Raises:
+ ConfigurationError: If credentials file is not found during refresh.
+ AuthenticationError: If OAuth token refresh or authentication fails.
"""
- if self.service is None:
- logger.debug('Connecting to Gmail service for %s', self.identifier)
- self.service = build('gmail', 'v1', credentials=self.creds, cache_discovery=False)
+ if self.service is not None:
+ return
+
+ if not self.creds.valid:
+ self.creds.refresh(Request()) # type: ignore[no-untyped-call]
+
+ logger.debug('Connecting to Gmail service for %s', self.identifier)
+ self.service = build('gmail', 'v1', credentials=self.creds, cache_discovery=False)
+ try:
+ self._label_map = {label['name']: label['id'] for label in self.list_labels()}
+ except Exception:
+ self.service = None
+ self._label_map = None
+ raise
def _api(self) -> Any:
"""Return the connected Gmail API service, connecting first if needed.
@@ -163,8 +109,8 @@ class GmailTarget:
"""
# Translate label names to their corresponding IDs
if self._label_map is None:
- # Get all labels from Gmail
- self._label_map = {label['name']: label['id'] for label in self.list_labels()}
+ self.connect()
+ assert self._label_map is not None
translated: list[str] = []
for label in labels:
label_id = self._label_map.get(label, None)
@@ -197,8 +143,6 @@ class GmailTarget:
RemoteError: If the Gmail API call fails.
"""
try:
- import base64
-
msg = RawMessage(raw_message)
encoded_message = base64.urlsafe_b64encode(msg.as_bytes(feed_name, delivery_name)).decode()
message_body: dict[str, Any] = {'raw': encoded_message}
@@ -215,33 +159,77 @@ class GmailTarget:
except HttpError as error:
raise RemoteError(f'An error occurred: {error}') from error
- @property
- def needs_auth(self) -> bool:
- """Check if this target needs re-authentication."""
- return self._needs_auth
-
def reauthenticate(self) -> None:
"""Perform OAuth re-authentication flow.
- Opens a browser window for the user to log in and authorize the app.
- Updates credentials upon successful authentication.
+ Delegates interactive re-authentication to the underlying credentials
+ provider and clears cached API state so the next operation reconnects
+ with the updated credentials.
Raises:
ConfigurationError: If credentials file is not found.
+ AuthenticationError: If interactive re-authentication is unsupported
+ or the OAuth flow fails.
"""
- if not self._credentials_file.exists():
- raise ConfigurationError(
- f'{self._credentials_file} not found. Please download it from Google Cloud Console.'
+ reauth = getattr(self.creds, 'reauthenticate', None)
+ if not callable(reauth):
+ raise AuthenticationError(
+ f"Target '{self.identifier}' does not support interactive reauthentication.",
+ target_id=self.identifier,
+ target_type='gmail',
)
-
- logger.info('Starting re-authentication for Gmail account %s', self.identifier)
-
- flow = InstalledAppFlow.from_client_secrets_file(str(self._credentials_file), SCOPES)
- self.creds = flow.run_local_server(port=0)
-
- # Save the credentials
- self._token_file.write_text(self.creds.to_json())
-
- self._needs_auth = False
- self.service = None # Reset service to force reconnect with new creds
- logger.info('Re-authentication successful for %s', self.identifier)
+ reauth()
+ self.service = None
+ self._label_map = None
+
+
+def make_gmail_credentials(
+ identifier: str,
+ details: Mapping[str, Any],
+ interactive: bool = True,
+ config_dir: Path | str | None = None,
+) -> BaseCredentials:
+ """Create credentials for a Gmail target from configuration.
+
+ Args:
+ identifier: Unique identifier for the Gmail target.
+ details: Target configuration mapping from ``korgalore.toml``.
+ interactive: Whether interactive OAuth browser flows are permitted.
+ config_dir: Optional base configuration directory for default token paths.
+
+ Returns:
+ Configured credentials instance compatible with ``GmailTarget``.
+
+ Raises:
+ ConfigurationError: If required configuration keys are missing or have
+ invalid types.
+ """
+ if 'credentials' not in details:
+ raise ConfigurationError(f"No credentials file specified for Gmail target '{identifier}'.")
+
+ creds_raw = details['credentials']
+ if not isinstance(creds_raw, str):
+ raise ConfigurationError(f"Gmail target '{identifier}': 'credentials' must be a string.")
+
+ token_raw = details.get('token')
+ if token_raw is not None and not isinstance(token_raw, str):
+ raise ConfigurationError(f"Gmail target '{identifier}': 'token' must be a string.")
+
+ if not token_raw:
+ base_dir = (
+ Path(config_dir)
+ if config_dir is not None
+ else Path(os.environ.get('XDG_CONFIG_HOME', Path.home() / '.config')) / 'korgalore'
+ )
+ token_file: str | Path = base_dir / f'gmail-{identifier}-token.json'
+ else:
+ token_file = token_raw
+
+ return OAuthFileCredentials(
+ identifier=identifier,
+ credentials_file=creds_raw,
+ token_file=token_file,
+ scopes=SCOPES,
+ interactive=interactive,
+ target_type='gmail',
+ )
diff --git a/src/korgalore/google_creds.py b/src/korgalore/google_creds.py
new file mode 100644
index 0000000..f1963bb
--- /dev/null
+++ b/src/korgalore/google_creds.py
@@ -0,0 +1,229 @@
+"""Google credential providers for korgalore."""
+
+import logging
+import os
+from collections.abc import Sequence
+from pathlib import Path
+from typing import Any
+
+from google.auth.credentials import Credentials as BaseCredentials
+from google.auth.exceptions import RefreshError
+from google.auth.external_account_authorized_user import Credentials as ExternalAccountCredentials
+from google.auth.transport.requests import Request
+from google.oauth2.credentials import Credentials
+from google_auth_oauthlib.flow import InstalledAppFlow # type: ignore[import-untyped]
+
+from korgalore import AuthenticationError, ConfigurationError
+
+logger = logging.getLogger('korgalore')
+
+# InstalledAppFlow.run_local_server() returns either flavour depending on what
+# the client_secrets file describes. Both expose the valid/expired/refresh_token
+# attributes and the to_json() and refresh() methods used below, so the code
+# does not care which one it gets -- but the declared type has to allow both.
+OAuthCredentials = Credentials | ExternalAccountCredentials
+
+
+class OAuthFileCredentials(BaseCredentials):
+ """Credentials backed by Google OAuth client secrets and a token file."""
+
+ def __init__(
+ self,
+ identifier: str,
+ credentials_file: str,
+ token_file: str | Path,
+ scopes: Sequence[str],
+ interactive: bool = True,
+ target_type: str = 'gmail',
+ ) -> None:
+ """Initialize file-backed OAuth credentials for a Google target.
+
+ Args:
+ identifier: Unique identifier for this target.
+ credentials_file: Path to the Google OAuth credentials JSON file.
+ token_file: Path to store/load the OAuth token.
+ scopes: OAuth scopes to request when running interactive auth.
+ interactive: If True, run OAuth flow interactively when needed.
+ If False, record ``needs_auth`` instead (for GUI mode).
+ target_type: Target type string reported in AuthenticationError.
+
+ Raises:
+ ConfigurationError: If credentials file path is empty or missing.
+ AuthenticationError: If token is expired/revoked and re-auth is needed,
+ or if an interactive OAuth flow fails.
+ """
+ self._creds: OAuthCredentials | None = None
+ super().__init__() # type: ignore[no-untyped-call]
+ self.identifier = identifier
+ self._interactive = interactive
+ self._scopes = list(scopes)
+ self._target_type = target_type
+ self.needs_auth = False
+
+ if not credentials_file:
+ raise ConfigurationError(f"No credentials file specified for target '{identifier}'.")
+ # Store expanded paths for potential re-authentication
+ self._credentials_file = Path(os.path.expandvars(credentials_file)).expanduser()
+ self._token_file = Path(os.path.expandvars(os.fspath(token_file))).expanduser()
+
+ self._load_credentials()
+
+ def _load_credentials(self) -> None:
+ """Load or refresh OAuth credentials for API access.
+
+ Attempts to load existing credentials from ``self._token_file``. If
+ not present or expired, refreshes with the stored refresh token or
+ initiates the OAuth flow using ``self._credentials_file``.
+
+ Raises:
+ ConfigurationError: If credentials file is not found.
+ AuthenticationError: If token is expired/revoked and re-auth is needed.
+ """
+ # The file token.json stores the user's access and refresh tokens
+ if self._token_file.exists():
+ try:
+ self._creds = Credentials.from_authorized_user_file(str(self._token_file)) # type: ignore[no-untyped-call]
+ except Exception as e:
+ logger.warning('Failed to load OAuth token for %s from %s: %s', self.identifier, self._token_file, e)
+ self._creds = None
+
+ # If there are no (valid) credentials available, let the user log in
+ if not self._creds or not self._creds.valid:
+ if self._creds and self._creds.expired and self._creds.refresh_token:
+ try:
+ self._creds.refresh(Request()) # type: ignore[no-untyped-call]
+ self._save_token()
+ except RefreshError as e:
+ logger.warning('OAuth token for target %s has expired or been revoked.', self.identifier)
+ self._quarantine_token()
+ self.needs_auth = True
+ if not self._interactive:
+ # In non-interactive mode, just return - caller will check needs_auth
+ return
+ raise AuthenticationError(
+ f"OAuth token for '{self.identifier}' is invalid. Please re-authenticate.",
+ target_id=self.identifier,
+ target_type=self._target_type,
+ ) from e
+ elif not self._credentials_file.exists():
+ raise ConfigurationError(
+ f"Credentials file not found for target '{self.identifier}': {self._credentials_file}"
+ )
+ elif not self._interactive:
+ # In non-interactive mode (GUI), don't run OAuth flow
+ # Just mark as needing auth and return - caller will check needs_auth
+ self.needs_auth = True
+ return
+ else:
+ self.reauthenticate()
+ return
+
+ self.needs_auth = False
+
+ @property
+ def token(self) -> str | None:
+ """Current OAuth2 access token, or None if not loaded."""
+ if self._creds and self._creds.token is not None:
+ return str(self._creds.token)
+ return None
+
+ @token.setter
+ def token(self, value: str | None) -> None:
+ if self._creds:
+ self._creds.token = value
+
+ @property
+ def expiry(self) -> Any:
+ """Expiration timestamp of the current access token, or None."""
+ return self._creds.expiry if self._creds else None
+
+ @expiry.setter
+ def expiry(self, value: Any) -> None:
+ if self._creds:
+ self._creds.expiry = value
+
+ @property
+ def valid(self) -> bool:
+ """Whether loaded OAuth2 credentials are present and currently valid."""
+ return bool(self._creds and self._creds.valid)
+
+ def _save_token(self) -> None:
+ """Persist current OAuth credentials to the configured token file."""
+ if self._creds is None:
+ return
+ # Save the credentials for the next run
+ self._token_file.parent.mkdir(parents=True, exist_ok=True)
+ self._token_file.write_text(self._creds.to_json()) # type: ignore[no-untyped-call]
+ self._token_file.chmod(0o600)
+
+ def _quarantine_token(self) -> None:
+ """Rename a revoked or invalid token file aside to ``.invalid``."""
+ if self._token_file.exists():
+ invalid_token_file = self._token_file.with_name(f'{self._token_file.name}.invalid')
+ invalid_token_file.unlink(missing_ok=True)
+ self._token_file.rename(invalid_token_file)
+
+ def refresh(self, request: Any = None) -> None:
+ """Refresh the access token using the stored refresh token.
+
+ Args:
+ request: Optional Google auth transport request object.
+
+ Raises:
+ ConfigurationError: If credentials file is missing during
+ interactive re-authentication.
+ AuthenticationError: If refresh fails or no token is loaded in
+ non-interactive mode.
+ """
+ if self._creds is None:
+ if self._interactive:
+ self.reauthenticate()
+ return
+ self.needs_auth = True
+ raise AuthenticationError(
+ f"Target '{self.identifier}' requires authentication.",
+ target_id=self.identifier,
+ target_type=self._target_type,
+ )
+ req = request or Request()
+ try:
+ self._creds.refresh(req) # type: ignore[no-untyped-call]
+ except RefreshError as e:
+ logger.warning('OAuth token for target %s has expired or been revoked.', self.identifier)
+ self._quarantine_token()
+ self.needs_auth = True
+ raise AuthenticationError(
+ f"Failed to refresh credentials for target '{self.identifier}': {e}",
+ target_id=self.identifier,
+ target_type=self._target_type,
+ ) from e
+ self._save_token()
+ self.needs_auth = False
+
+ def reauthenticate(self) -> None:
+ """Perform OAuth re-authentication flow.
+
+ Opens a browser window for the user to log in and authorize the app.
+ Updates credentials upon successful authentication.
+
+ Raises:
+ ConfigurationError: If credentials file is not found.
+ AuthenticationError: If the local OAuth authorization flow fails.
+ """
+ if not self._credentials_file.exists():
+ raise ConfigurationError(
+ f"Credentials file not found for target '{self.identifier}': {self._credentials_file}"
+ )
+ logger.critical('Log in to account for %s', self.identifier)
+ try:
+ flow = InstalledAppFlow.from_client_secrets_file(str(self._credentials_file), self._scopes)
+ self._creds = flow.run_local_server(port=0)
+ except Exception as e:
+ self.needs_auth = True
+ raise AuthenticationError(
+ f"OAuth flow failed for target '{self.identifier}': {e}",
+ target_id=self.identifier,
+ target_type=self._target_type,
+ ) from e
+ self._save_token()
+ self.needs_auth = False
diff --git a/tests/test_gmail_target.py b/tests/test_gmail_target.py
index c3a3ed4..9e27f1c 100644
--- a/tests/test_gmail_target.py
+++ b/tests/test_gmail_target.py
@@ -5,28 +5,25 @@ from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
-from google.auth.exceptions import RefreshError
+from google.auth.credentials import Credentials as BaseCredentials
from googleapiclient.errors import HttpError # type: ignore[import-untyped]
-from korgalore import ConfigurationError, RemoteError
-from korgalore.gmail_target import SCOPES, GmailTarget
+from korgalore import AuthenticationError, ConfigurationError, RemoteError
+from korgalore.gmail_target import SCOPES, GmailTarget, make_gmail_credentials
+from korgalore.google_creds import OAuthFileCredentials
-def make_target(tmp_path: Path) -> tuple[GmailTarget, MagicMock]:
- """A GmailTarget with valid stored credentials; returns it with those credentials."""
- token = tmp_path / 'token.json'
- token.write_text('{}')
- with patch('korgalore.gmail_target.Credentials') as mock_creds_class:
- mock_creds = MagicMock()
- mock_creds.valid = True
- mock_creds_class.from_authorized_user_file.return_value = mock_creds
- return GmailTarget('test', str(tmp_path / 'creds.json'), str(token)), mock_creds
+def make_target() -> tuple[GmailTarget, MagicMock]:
+ """A GmailTarget with valid mock credentials."""
+ mock_creds = MagicMock(spec=BaseCredentials)
+ mock_creds.valid = True
+ return GmailTarget('test', creds=mock_creds), mock_creds
@pytest.fixture
-def gmail(tmp_path: Path) -> tuple[GmailTarget, MagicMock]:
+def gmail() -> tuple[GmailTarget, MagicMock]:
"""A target with a mocked API service; the label map is not loaded yet."""
- target, _ = make_target(tmp_path)
+ target, _ = make_target()
service = MagicMock()
target.service = service
return target, service
@@ -52,123 +49,85 @@ def import_body(service: MagicMock) -> dict[str, object]:
class TestGmailTargetInit:
- """Tests for GmailTarget initialization."""
+ """Tests for GmailTarget initialization and credential delegation."""
- @patch('korgalore.gmail_target.Credentials')
- def test_loads_existing_valid_token(self, mock_credentials: MagicMock, tmp_path: Path) -> None:
- """Loads credentials from the stored token; the service waits for connect()."""
- token = tmp_path / 'token.json'
- token.write_text('{}')
- mock_creds = MagicMock()
- mock_creds.valid = True
- mock_credentials.from_authorized_user_file.return_value = mock_creds
-
- target = GmailTarget('test', str(tmp_path / 'creds.json'), str(token))
+ def test_init_with_credentials(self) -> None:
+ """Stores identifier and credentials without building API service yet."""
+ target, mock_creds = make_target()
assert target.identifier == 'test'
assert target.creds is mock_creds
assert target.service is None
- mock_credentials.from_authorized_user_file.assert_called_once_with(str(token), SCOPES)
-
- @patch('korgalore.gmail_target.Credentials')
- @patch('korgalore.gmail_target.Request')
- def test_refreshes_expired_token(
- self, mock_request: MagicMock, mock_credentials: MagicMock, tmp_path: Path
- ) -> None:
- """Refreshes expired credentials with refresh token and saves the result."""
- token = tmp_path / 'token.json'
- token.write_text('{"token": "stale"}')
- mock_creds = MagicMock()
- mock_creds.valid = False
- mock_creds.expired = True
- mock_creds.refresh_token = 'refresh_token_value'
- mock_creds.to_json.return_value = '{"token": "refreshed"}'
- mock_credentials.from_authorized_user_file.return_value = mock_creds
-
- GmailTarget('test', str(tmp_path / 'creds.json'), str(token))
-
- mock_creds.refresh.assert_called_once()
- assert token.read_text() == '{"token": "refreshed"}'
- @patch('korgalore.gmail_target.Credentials')
- @patch('korgalore.gmail_target.Request')
- def test_revoked_token_is_set_aside(
- self, mock_request: MagicMock, mock_credentials: MagicMock, tmp_path: Path
- ) -> None:
- """A token that fails to refresh is renamed out of the way and auth is flagged."""
- token = tmp_path / 'token.json'
- token.write_text('{"token": "revoked"}')
- invalid = tmp_path / 'token.json.invalid'
- invalid.write_text('{"token": "older"}')
- mock_creds = MagicMock()
- mock_creds.valid = False
- mock_creds.expired = True
- mock_creds.refresh_token = 'refresh_token_value'
- mock_creds.refresh.side_effect = RefreshError
- mock_credentials.from_authorized_user_file.return_value = mock_creds
-
- target = GmailTarget('test', str(tmp_path / 'creds.json'), str(token), interactive=False)
-
- assert target.needs_auth
- assert not token.exists()
- assert invalid.read_text() == '{"token": "revoked"}'
-
- @patch('korgalore.gmail_target.InstalledAppFlow')
- def test_runs_oauth_flow_and_saves_token_when_no_token(self, mock_flow_class: MagicMock, tmp_path: Path) -> None:
- """Runs the OAuth flow when no token exists, then saves the new token."""
- creds = tmp_path / 'creds.json'
- creds.write_text('{}')
- token = tmp_path / 'token.json'
- mock_creds = MagicMock()
- mock_creds.to_json.return_value = '{"access_token": "new_token"}'
- mock_flow = MagicMock()
- mock_flow.run_local_server.return_value = mock_creds
- mock_flow_class.from_client_secrets_file.return_value = mock_flow
-
- target = GmailTarget('test', str(creds), str(token))
-
- mock_flow_class.from_client_secrets_file.assert_called_once_with(str(creds), SCOPES)
- mock_flow.run_local_server.assert_called_once_with(port=0)
- assert target.creds is mock_creds
- assert token.read_text() == '{"access_token": "new_token"}'
+ def test_reauthenticate_delegates_and_resets_service(self, gmail_labels: tuple[GmailTarget, MagicMock]) -> None:
+ """Delegates reauthenticate() to creds and clears cached service state."""
+ target, _ = gmail_labels
+ reauth_mock = MagicMock()
+ setattr(target.creds, 'reauthenticate', reauth_mock)
- def test_missing_credentials_file_raises(self, tmp_path: Path) -> None:
- """Missing credentials file raises ConfigurationError."""
- # Neither the token file nor the credentials file exists
- with pytest.raises(ConfigurationError) as exc_info:
- GmailTarget('test', str(tmp_path / 'creds.json'), str(tmp_path / 'token.json'))
- assert 'not found' in str(exc_info.value)
- assert 'creds.json' in str(exc_info.value)
+ target.reauthenticate()
- @patch('korgalore.gmail_target.Credentials')
- def test_expands_user_paths(self, mock_credentials: MagicMock, tmp_path: Path) -> None:
- """Tilde and env vars in paths are expanded."""
- token = tmp_path / 'token.json'
- token.write_text('{}')
- mock_credentials.from_authorized_user_file.return_value = MagicMock(valid=True)
+ reauth_mock.assert_called_once_with()
+ assert target.service is None
+ assert target._label_map is None
- with patch.dict('os.environ', {'HOME': str(tmp_path)}):
- GmailTarget('test', '~/creds.json', '$HOME/token.json')
+ def test_reauthenticate_unsupported_raises(self) -> None:
+ """Raises AuthenticationError when credentials do not support reauth."""
+ target, _ = make_target()
- mock_credentials.from_authorized_user_file.assert_called_once_with(str(token), SCOPES)
+ with pytest.raises(AuthenticationError, match='does not support interactive reauthentication'):
+ target.reauthenticate()
class TestGmailTargetConnect:
"""Tests for GmailTarget connect method."""
@patch('korgalore.gmail_target.build')
- def test_connect_builds_service_once(self, mock_build: MagicMock, tmp_path: Path) -> None:
+ def test_connect_builds_service_once(self, mock_build: MagicMock) -> None:
"""Connect builds the Gmail API service, and repeated calls don't rebuild it."""
- target, mock_creds = make_target(tmp_path)
+ target, mock_creds = make_target()
mock_service = MagicMock()
+ mock_service.users().labels().list().execute.return_value = {'labels': [{'id': 'INBOX', 'name': 'INBOX'}]}
mock_build.return_value = mock_service
target.connect()
+ mock_build.assert_called_once_with('gmail', 'v1', credentials=mock_creds, cache_discovery=False)
+ assert target.service is mock_service
+ assert target._label_map == {'INBOX': 'INBOX'}
+
target.connect()
target.connect()
+ mock_build.assert_called_once()
- mock_build.assert_called_once_with('gmail', 'v1', credentials=mock_creds, cache_discovery=False)
- assert target.service is mock_service
+ @patch('korgalore.gmail_target.build')
+ def test_connect_failure_resets_state(self, mock_build: MagicMock) -> None:
+ """Label probe failure resets service and label map to None."""
+ target, _ = make_target()
+ mock_service = MagicMock()
+ mock_service.users().labels().list().execute.side_effect = http_error(403, b'Forbidden')
+ mock_build.return_value = mock_service
+
+ with pytest.raises(RemoteError):
+ target.connect()
+
+ assert target.service is None
+ assert target._label_map is None
+
+ @patch('korgalore.gmail_target.build')
+ @patch('korgalore.gmail_target.Request')
+ def test_connect_refreshes_invalid_credentials(self, mock_request: MagicMock, mock_build: MagicMock) -> None:
+ """Refreshes invalid credentials before building the API service."""
+ target, mock_creds = make_target()
+ mock_creds.valid = False
+ mock_service = MagicMock()
+ mock_service.users().labels().list().execute.return_value = {'labels': []}
+ mock_build.return_value = mock_service
+
+ target.connect()
+
+ mock_creds.refresh.assert_called_once_with(mock_request.return_value)
+ mock_build.assert_called_once()
+ assert target._label_map == {}
class TestGmailTargetListLabels:
@@ -215,9 +174,12 @@ class TestGmailTargetListLabels:
class TestGmailTargetTranslateLabels:
"""Tests for GmailTarget translate_labels method."""
- def test_translate_labels(self, gmail: tuple[GmailTarget, MagicMock]) -> None:
+ @patch('korgalore.gmail_target.build')
+ def test_translate_labels(self, mock_build: MagicMock) -> None:
"""Label names become IDs, in order; the label map is fetched only once."""
- target, service = gmail
+ target, _ = make_target()
+ service = MagicMock()
+ mock_build.return_value = service
service.users().labels().list().execute.return_value = {
'labels': [
{'id': 'INBOX', 'name': 'INBOX'},
@@ -231,9 +193,13 @@ class TestGmailTargetTranslateLabels:
assert target.translate_labels(['MyLabel']) == ['Label_123']
assert service.users().labels().list().execute.call_count == 1
+ assert mock_build.call_count == 1
- def test_translate_unknown_label_raises(self, gmail: tuple[GmailTarget, MagicMock]) -> None:
- target, service = gmail
+ @patch('korgalore.gmail_target.build')
+ def test_translate_unknown_label_raises(self, mock_build: MagicMock) -> None:
+ target, _ = make_target()
+ service = MagicMock()
+ mock_build.return_value = service
service.users().labels().list().execute.return_value = {'labels': [{'id': 'INBOX', 'name': 'INBOX'}]}
with pytest.raises(ConfigurationError) as exc_info:
@@ -241,8 +207,8 @@ class TestGmailTargetTranslateLabels:
assert 'not found' in str(exc_info.value)
assert 'NonExistent' in str(exc_info.value)
- def test_label_names_are_case_sensitive(self, gmail: tuple[GmailTarget, MagicMock]) -> None:
- target, _ = gmail
+ def test_label_names_are_case_sensitive(self) -> None:
+ target, _ = make_target()
target._label_map = {'INBOX': 'INBOX', 'inbox': 'inbox_lower'}
assert target.translate_labels(['INBOX']) == ['INBOX']
@@ -329,3 +295,40 @@ class TestGmailTargetImportMessage:
results = [target.import_message(f'Message {i}'.encode(), ['INBOX']) for i in range(5)]
assert [r['id'] for r in results] == ['msg0', 'msg1', 'msg2', 'msg3', 'msg4']
+
+
+class TestMakeGmailCredentials:
+ """Tests for make_gmail_credentials configuration factory."""
+
+ def test_no_credentials_raises(self) -> None:
+ """Omitting credentials raises ConfigurationError."""
+ with pytest.raises(ConfigurationError, match='No credentials file specified'):
+ make_gmail_credentials('test', {})
+
+ @pytest.mark.parametrize(
+ ('details', 'expected_match'),
+ [
+ ({'credentials': 123}, "'credentials' must be a string"),
+ ({'credentials': '/tmp/creds.json', 'token': 456}, "'token' must be a string"),
+ ],
+ ids=['non-str-credentials', 'non-str-token'],
+ )
+ def test_invalid_option_types_raise(self, details: dict[str, object], expected_match: str) -> None:
+ """Non-string configuration values raise ConfigurationError."""
+ with pytest.raises(ConfigurationError, match=expected_match):
+ make_gmail_credentials('test', details)
+
+ @patch('korgalore.google_creds.Credentials')
+ def test_default_token_file_uses_xdg_config_dir(self, mock_credentials: MagicMock, tmp_path: Path) -> None:
+ """When token is omitted, defaults to gmail-{identifier}-token.json in XDG config dir."""
+ default_token = tmp_path / 'korgalore' / 'gmail-personal-token.json'
+ default_token.parent.mkdir(parents=True)
+ default_token.write_text('{}')
+ mock_credentials.from_authorized_user_file.return_value = MagicMock(valid=True)
+
+ with patch.dict('os.environ', {'XDG_CONFIG_HOME': str(tmp_path)}):
+ creds = make_gmail_credentials('personal', {'credentials': str(tmp_path / 'creds.json')})
+
+ mock_credentials.from_authorized_user_file.assert_called_once_with(str(default_token))
+ assert isinstance(creds, OAuthFileCredentials)
+ assert creds._scopes == SCOPES
diff --git a/tests/test_google_creds.py b/tests/test_google_creds.py
new file mode 100644
index 0000000..4ac0a3d
--- /dev/null
+++ b/tests/test_google_creds.py
@@ -0,0 +1,154 @@
+"""Tests for Google OAuth2 credential providers."""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+from google.auth.exceptions import RefreshError
+
+from korgalore import AuthenticationError, ConfigurationError
+from korgalore.google_creds import OAuthFileCredentials
+
+TEST_SCOPES = ['https://example.invalid/scope.a', 'https://example.invalid/scope.b']
+
+
+class TestOAuthFileCredentials:
+ """Tests for OAuthFileCredentials initialization, refresh, and reauth."""
+
+ @patch('korgalore.google_creds.Credentials')
+ def test_loads_existing_valid_token(self, mock_credentials: MagicMock, tmp_path: Path) -> None:
+ """Loads credentials from the stored token when already valid."""
+ token = tmp_path / 'token.json'
+ token.write_text('{}')
+ mock_creds = MagicMock()
+ mock_creds.valid = True
+ mock_creds.token = 'access-tok'
+ mock_credentials.from_authorized_user_file.return_value = mock_creds
+
+ provider = OAuthFileCredentials('test', str(tmp_path / 'creds.json'), token, scopes=TEST_SCOPES)
+
+ assert provider.valid
+ assert provider.token == 'access-tok'
+ assert not provider.needs_auth
+ mock_credentials.from_authorized_user_file.assert_called_once_with(str(token))
+
+ @patch('korgalore.google_creds.Credentials')
+ @patch('korgalore.google_creds.Request')
+ def test_refreshes_expired_token(
+ self, mock_request: MagicMock, mock_credentials: MagicMock, tmp_path: Path
+ ) -> None:
+ """Refreshes expired credentials with refresh token and saves the result."""
+ token = tmp_path / 'token.json'
+ token.write_text('{"token": "stale"}')
+ mock_creds = MagicMock()
+ mock_creds.valid = False
+ mock_creds.expired = True
+ mock_creds.refresh_token = 'refresh_token_value'
+ mock_creds.to_json.return_value = '{"token": "refreshed"}'
+ mock_credentials.from_authorized_user_file.return_value = mock_creds
+
+ OAuthFileCredentials('test', str(tmp_path / 'creds.json'), token, scopes=TEST_SCOPES)
+
+ mock_creds.refresh.assert_called_once()
+ assert token.read_text() == '{"token": "refreshed"}'
+ assert (token.stat().st_mode & 0o777) == 0o600
+
+ @patch('korgalore.google_creds.Credentials')
+ @patch('korgalore.google_creds.Request')
+ def test_revoked_token_is_set_aside(
+ self, mock_request: MagicMock, mock_credentials: MagicMock, tmp_path: Path
+ ) -> None:
+ """A token that fails to refresh is renamed aside and needs_auth is flagged."""
+ token = tmp_path / 'token.json'
+ token.write_text('{"token": "revoked"}')
+ invalid = tmp_path / 'token.json.invalid'
+ invalid.write_text('{"token": "older"}')
+ mock_creds = MagicMock()
+ mock_creds.valid = False
+ mock_creds.expired = True
+ mock_creds.refresh_token = 'refresh_token_value'
+ mock_creds.refresh.side_effect = RefreshError
+ mock_credentials.from_authorized_user_file.return_value = mock_creds
+
+ provider = OAuthFileCredentials(
+ 'test', str(tmp_path / 'creds.json'), token, scopes=TEST_SCOPES, interactive=False
+ )
+
+ assert provider.needs_auth
+ assert not token.exists()
+ assert invalid.read_text() == '{"token": "revoked"}'
+
+ @patch('korgalore.google_creds.Credentials')
+ @patch('korgalore.google_creds.Request')
+ def test_revoked_token_interactive_raises_authentication_error(
+ self, mock_request: MagicMock, mock_credentials: MagicMock, tmp_path: Path
+ ) -> None:
+ """Interactive load raises AuthenticationError when token refresh is revoked."""
+ token = tmp_path / 'token.json'
+ token.write_text('{"token": "revoked"}')
+ mock_creds = MagicMock()
+ mock_creds.valid = False
+ mock_creds.expired = True
+ mock_creds.refresh_token = 'refresh_token_value'
+ mock_creds.refresh.side_effect = RefreshError
+ mock_credentials.from_authorized_user_file.return_value = mock_creds
+
+ with pytest.raises(AuthenticationError, match='invalid'):
+ OAuthFileCredentials('test', str(tmp_path / 'creds.json'), token, scopes=TEST_SCOPES, interactive=True)
+
+ assert not token.exists()
+ assert token.with_name('token.json.invalid').exists()
+
+ @patch('korgalore.google_creds.InstalledAppFlow')
+ def test_runs_oauth_flow_and_saves_token_when_no_token(self, mock_flow_class: MagicMock, tmp_path: Path) -> None:
+ """Runs the OAuth flow interactively, saves the returned credentials, and marks auth done."""
+ creds = tmp_path / 'creds.json'
+ creds.write_text('{}')
+ token = tmp_path / 'token.json'
+ mock_creds = MagicMock()
+ mock_creds.valid = True
+ mock_creds.to_json.return_value = '{"token": "fresh"}'
+ mock_flow = MagicMock()
+ mock_flow.run_local_server.return_value = mock_creds
+ mock_flow_class.from_client_secrets_file.return_value = mock_flow
+
+ provider = OAuthFileCredentials('test', str(creds), token, scopes=TEST_SCOPES)
+
+ mock_flow_class.from_client_secrets_file.assert_called_once_with(str(creds), TEST_SCOPES)
+ mock_flow.run_local_server.assert_called_once_with(port=0)
+ assert provider.valid
+ assert not provider.needs_auth
+ assert token.read_text() == '{"token": "fresh"}'
+ assert (token.stat().st_mode & 0o777) == 0o600
+
+ def test_missing_credentials_file_raises(self, tmp_path: Path) -> None:
+ """Missing credentials file raises ConfigurationError."""
+ with pytest.raises(ConfigurationError) as exc_info:
+ OAuthFileCredentials('test', str(tmp_path / 'creds.json'), tmp_path / 'token.json', scopes=TEST_SCOPES)
+ assert 'not found' in str(exc_info.value)
+ assert 'creds.json' in str(exc_info.value)
+
+ @patch('korgalore.google_creds.Credentials')
+ def test_expands_user_paths(self, mock_credentials: MagicMock, tmp_path: Path) -> None:
+ """Tilde and environment variables in paths are expanded."""
+ token = tmp_path / 'token.json'
+ token.write_text('{}')
+ creds = tmp_path / 'creds.json'
+ creds.write_text('{}')
+ mock_credentials.from_authorized_user_file.return_value = MagicMock(valid=True)
+
+ with patch.dict('os.environ', {'HOME': str(tmp_path)}):
+ OAuthFileCredentials('test', '~/creds.json', '$HOME/token.json', scopes=TEST_SCOPES)
+
+ mock_credentials.from_authorized_user_file.assert_called_once_with(str(token))
+
+ def test_corrupt_token_file_falls_back_gracefully(self, tmp_path: Path) -> None:
+ """Corrupt or truncated token.json logs warning and leaves creds unloaded instead of crashing."""
+ token = tmp_path / 'token.json'
+ token.write_text('not valid json{')
+ creds = tmp_path / 'creds.json'
+ creds.write_text('{}')
+
+ provider = OAuthFileCredentials('test', str(creds), token, scopes=TEST_SCOPES, interactive=False)
+ assert provider.needs_auth
+ assert not provider.valid
--
2.56.0.542.g3ef89b7020-goog
^ permalink raw reply related [flat|nested] 4+ messages in thread* [PATCH kgl 2/2] Add token_command support for Gmail targets
2026-11-11 6:59 [PATCH kgl 0/2] Add token_command support for Gmail targets Dmitry Torokhov
2026-11-11 6:59 ` [PATCH kgl 1/2] Decouple GmailTarget from OAuth file credential management Dmitry Torokhov
@ 2026-11-11 6:59 ` Dmitry Torokhov
2026-11-12 19:58 ` [PATCH kgl 0/2] " Konstantin Ryabitsev
2 siblings, 0 replies; 4+ messages in thread
From: Dmitry Torokhov @ 2026-11-11 6:59 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev
Add support for obtaining Gmail OAuth access tokens from an external
command via the token_command configuration setting on Gmail targets.
This enables integrating with system token managers, secret stores, or
command-line helpers without needing Google OAuth client secrets or
running interactive OAuth login flows in a browser.
Signed-off-by: Dmitry Torokhov <dmitry.torokhov@gmail.com>
---
docs/configuration.rst | 10 +++-
korgalore-example.toml | 2 +
man/korgalore.toml.5 | 12 ++++-
src/korgalore/cli.py | 12 ++++-
src/korgalore/gmail_target.py | 27 +++++++---
src/korgalore/google_creds.py | 111 +++++++++++++++++++++++++++++++++++++++-
tests/test_cli_misc.py | 103 +++++++++++++++++++++++++++++++++++++-
tests/test_gmail_target.py | 35 +++++++++++--
tests/test_google_creds.py | 114 ++++++++++++++++++++++++++++++++++++++++--
9 files changed, 404 insertions(+), 22 deletions(-)
diff --git a/docs/configuration.rst b/docs/configuration.rst
index fbba776..734ab26 100644
--- a/docs/configuration.rst
+++ b/docs/configuration.rst
@@ -410,9 +410,17 @@ Gmail Target Parameters
~~~~~~~~~~~~~~~~~~~~~~~~
* ``type``: Must be ``'gmail'``
-* ``credentials``: Path to the OAuth credentials JSON file
+* ``credentials``: Path to the OAuth credentials JSON file (required unless
+ ``token_command`` is specified)
* ``token``: (Optional) Path to store the OAuth token. Defaults to
``~/.config/korgalore/gmail-{identifier}-token.json``
+* ``token_command``: (Optional) External command that prints a valid OAuth2
+ bearer token to stdout. Mutually exclusive with both ``credentials`` and
+ ``token``. The command string is split using standard command-line syntax
+ and executed directly without a shell, with tilde and environment variable
+ expansion. The bearer token returned by the command must carry the OAuth
+ scopes required by korgalore (``https://www.googleapis.com/auth/gmail.labels``
+ and ``https://www.googleapis.com/auth/gmail.insert``).
Maildir Target Parameters
~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/korgalore-example.toml b/korgalore-example.toml
index b7d8164..0551bc4 100644
--- a/korgalore-example.toml
+++ b/korgalore-example.toml
@@ -3,6 +3,8 @@
[targets.personal]
type = 'gmail'
credentials = '/path/to/personal-credentials.json'
+# Alternatively, obtain tokens from an external command:
+# token_command = 'oama access personal'
[targets.work]
type = 'gmail'
diff --git a/man/korgalore.toml.5 b/man/korgalore.toml.5
index f72d242..3d23a74 100644
--- a/man/korgalore.toml.5
+++ b/man/korgalore.toml.5
@@ -30,11 +30,21 @@ name and a type that determines which parameters are required.
Must be \fI'gmail'\fR.
.TP
.B credentials
-Path to the OAuth credentials JSON file from Google Cloud Console.
+Path to the OAuth credentials JSON file from Google Cloud Console (required
+unless \fBtoken_command\fR is specified).
.TP
.B token
(Optional) Path to store the OAuth token. Defaults to
\fI~/.config/korgalore/gmail-<name>-token.json\fR.
+.TP
+.B token_command
+(Optional) External command to execute that prints an OAuth2 bearer token to
+stdout. Mutually exclusive with both \fBcredentials\fR and \fBtoken\fR.
+The command is split into arguments using standard command-line syntax and
+executed directly without a shell, with tilde and environment variable
+expansion. The bearer token must carry the OAuth scopes required by
+korgalore (\fIhttps://www.googleapis.com/auth/gmail.labels\fR and
+\fIhttps://www.googleapis.com/auth/gmail.insert\fR).
.RE
.PP
.B Maildir Targets
diff --git a/src/korgalore/cli.py b/src/korgalore/cli.py
index 180906d..e98d4be 100644
--- a/src/korgalore/cli.py
+++ b/src/korgalore/cli.py
@@ -1948,6 +1948,9 @@ def auth(ctx: click.Context, target: str | None) -> None:
ts = get_target(ctx, identifier)
ts.connect()
logger.info('Authenticated target: %s (type: %s)', identifier, target_type)
+ except (ConfigurationError, AuthenticationError, RemoteError) as e:
+ logger.error('Failed to authenticate target %s: %s', identifier, e)
+ raise click.Abort() from e
except click.Abort:
logger.error('Failed to authenticate target: %s', identifier)
raise
@@ -2483,7 +2486,12 @@ def yank(ctx: click.Context, target: str | None, labels: tuple[str, ...], thread
uploaded = 0
failed = 0
- ts.connect()
+ try:
+ ts.connect()
+ except (ConfigurationError, AuthenticationError, liblore.RemoteError) as e:
+ logger.critical('Failed to connect to target "%s": %s', target, e)
+ raise click.Abort() from e
+
with click.progressbar(
messages, label='Uploading thread', show_pos=True, file=progress_file(ctx.obj['hide_bar'])
) as bar:
@@ -2524,7 +2532,7 @@ def yank(ctx: click.Context, target: str | None, labels: tuple[str, ...], thread
ts.connect()
ts.import_message(raw_message, labels=labels_list)
logger.info('Successfully uploaded message.')
- except liblore.RemoteError as e:
+ except (ConfigurationError, AuthenticationError, liblore.RemoteError) as e:
logger.critical('Failed to upload message: %s', str(e))
raise click.Abort() from e
diff --git a/src/korgalore/gmail_target.py b/src/korgalore/gmail_target.py
index 3fd9f8b..868e148 100644
--- a/src/korgalore/gmail_target.py
+++ b/src/korgalore/gmail_target.py
@@ -11,7 +11,7 @@ from googleapiclient.discovery import Resource, build # type: ignore[import-unt
from googleapiclient.errors import HttpError # type: ignore[import-untyped]
from korgalore import AuthenticationError, ConfigurationError, RemoteError
-from korgalore.google_creds import OAuthFileCredentials
+from korgalore.google_creds import CommandCredentials, OAuthFileCredentials, parse_token_command
from korgalore.message import RawMessage
logger = logging.getLogger('korgalore')
@@ -53,8 +53,11 @@ class GmailTarget:
Creates the Gmail API service object if not already connected.
Raises:
- ConfigurationError: If credentials file is not found during refresh.
+ ConfigurationError: If credentials file is missing or token_command
+ cannot be executed.
AuthenticationError: If OAuth token refresh or authentication fails.
+ RemoteError: If token_command fails, times out, or returns an
+ invalid token.
"""
if self.service is not None:
return
@@ -201,11 +204,23 @@ def make_gmail_credentials(
Configured credentials instance compatible with ``GmailTarget``.
Raises:
- ConfigurationError: If required configuration keys are missing or have
- invalid types.
+ ConfigurationError: If required configuration keys are missing,
+ mutually exclusive, or invalid.
"""
- if 'credentials' not in details:
- raise ConfigurationError(f"No credentials file specified for Gmail target '{identifier}'.")
+ has_creds = 'credentials' in details
+ has_token = 'token' in details
+ has_cmd = 'token_command' in details
+
+ if has_cmd:
+ if has_creds or has_token:
+ raise ConfigurationError(
+ f"Gmail target '{identifier}' cannot specify both 'token_command' and 'credentials'/'token'."
+ )
+ command = parse_token_command(details['token_command'], f"Gmail target '{identifier}'")
+ return CommandCredentials(identifier=identifier, command=command)
+
+ if not has_creds:
+ raise ConfigurationError(f"Gmail target '{identifier}' must specify either 'credentials' or 'token_command'.")
creds_raw = details['credentials']
if not isinstance(creds_raw, str):
diff --git a/src/korgalore/google_creds.py b/src/korgalore/google_creds.py
index f1963bb..0818244 100644
--- a/src/korgalore/google_creds.py
+++ b/src/korgalore/google_creds.py
@@ -1,7 +1,9 @@
-"""Google credential providers for korgalore."""
+"""Google OAuth2 and command credential providers for korgalore."""
import logging
import os
+import shlex
+import subprocess
from collections.abc import Sequence
from pathlib import Path
from typing import Any
@@ -13,7 +15,7 @@ from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow # type: ignore[import-untyped]
-from korgalore import AuthenticationError, ConfigurationError
+from korgalore import AuthenticationError, ConfigurationError, RemoteError
logger = logging.getLogger('korgalore')
@@ -227,3 +229,108 @@ class OAuthFileCredentials(BaseCredentials):
) from e
self._save_token()
self.needs_auth = False
+
+
+def parse_token_command(command: Any, context: str) -> list[str]:
+ """Validate and parse a ``token_command`` value into an argument vector.
+
+ Args:
+ command: Raw configuration value expected to be a non-empty command string.
+ context: Human-readable caller label for error messages.
+
+ Returns:
+ Argument vector with tilde and environment variables expanded.
+
+ Raises:
+ ConfigurationError: If ``command`` is not a string, has invalid shell
+ quoting, or splits into an empty argument list.
+ """
+ if not isinstance(command, str):
+ raise ConfigurationError(f"{context}: 'token_command' must be a string.")
+ try:
+ raw_args = shlex.split(command)
+ except ValueError as e:
+ raise ConfigurationError(f"Invalid 'token_command' syntax for {context}: {e}") from e
+ if not raw_args:
+ raise ConfigurationError(f"{context}: 'token_command' cannot be empty.")
+ return [os.path.expanduser(os.path.expandvars(arg)) for arg in raw_args] # noqa: PTH111
+
+
+def run_token_command(command: Sequence[str], context: str, timeout: int = 30) -> str:
+ """Execute an external token helper and return its bearer token output.
+
+ Runs ``command`` directly without a shell and verifies that standard output
+ contains a single non-empty, whitespace-free token.
+
+ Args:
+ command: Argument vector passed directly to ``subprocess.run``.
+ context: Human-readable caller label for debug logs and errors.
+ timeout: Maximum execution time in seconds before failing.
+
+ Returns:
+ Stripped bearer token read from standard output.
+
+ Raises:
+ ConfigurationError: If the executable cannot be launched.
+ RemoteError: If the command times out, exits non-zero, or returns an
+ empty or whitespace-containing token.
+ """
+ try:
+ res = subprocess.run(
+ command,
+ stdin=subprocess.DEVNULL,
+ capture_output=True,
+ text=True,
+ encoding='utf-8',
+ errors='replace',
+ timeout=timeout,
+ check=False,
+ )
+ except subprocess.TimeoutExpired as e:
+ raise RemoteError(f'token_command timed out after {timeout}s for {context}') from e
+ except OSError as e:
+ raise ConfigurationError(f'Failed to execute token_command for {context}: {e}') from e
+
+ err_msg = ''
+ if res.stderr:
+ logger.debug('token_command stderr for %s:\n%s', context, res.stderr)
+ lines = [line.strip() for line in res.stderr.splitlines() if line.strip()]
+ if lines:
+ err_msg = f': {lines[-1]}'
+
+ if res.returncode != 0:
+ raise RemoteError(f'token_command for {context} failed with exit code {res.returncode}{err_msg}')
+
+ token = res.stdout.strip()
+ if not token or any(c.isspace() for c in token):
+ raise RemoteError(f'token_command for {context} returned empty or invalid token{err_msg}')
+
+ return token
+
+
+class CommandCredentials(BaseCredentials):
+ """Credentials obtained by executing an external token command."""
+
+ def __init__(self, identifier: str, command: Sequence[str]) -> None:
+ """Initialize command-backed bearer credentials.
+
+ Args:
+ identifier: Unique identifier for the target.
+ command: Parsed command argument vector to run on refresh.
+ """
+ super().__init__() # type: ignore[no-untyped-call]
+ self.identifier = identifier
+ self.command = list(command)
+
+ def refresh(self, request: Any = None) -> None:
+ """Run the configured token command and capture the bearer token.
+
+ Args:
+ request: Ignored; present to satisfy the ``google.auth`` interface.
+
+ Raises:
+ ConfigurationError: If the token command cannot be executed.
+ RemoteError: If the token command fails, times out, or emits invalid
+ output.
+ """
+ self.token = run_token_command(self.command, f"target '{self.identifier}'")
diff --git a/tests/test_cli_misc.py b/tests/test_cli_misc.py
index e351b0b..6951521 100644
--- a/tests/test_cli_misc.py
+++ b/tests/test_cli_misc.py
@@ -10,12 +10,16 @@ noise into somebody's debug log.
"""
import io
+import subprocess
+from unittest.mock import patch
import click
import pytest
+from click.testing import CliRunner
-from korgalore import format_key_for_display
-from korgalore.cli import progress_file
+from korgalore import RemoteError, format_key_for_display
+from korgalore.cli import auth, perform_yank, progress_file, yank
+from tests.digest_helpers import make_ctx
@pytest.mark.parametrize(
@@ -75,3 +79,98 @@ def test_hidden_bar_writes_nothing_to_stdout(capsys: pytest.CaptureFixture[str])
# stream is what keeps it off the user's terminal, so check it landed here.
assert isinstance(stream, io.StringIO)
assert 'Hiding' in stream.getvalue()
+
+
+def test_auth_token_command_failure_aborts_cleanly() -> None:
+ runner = CliRunner()
+ ctx_obj = {
+ 'config': {'targets': {'test': {'type': 'gmail', 'token_command': 'get-token'}}},
+ 'targets': {},
+ }
+ with patch('korgalore.google_creds.subprocess.run') as mock_run:
+ mock_run.return_value = subprocess.CompletedProcess(
+ args=['get-token'], returncode=1, stdout='', stderr='command failed'
+ )
+ res = runner.invoke(auth, ['test'], obj=ctx_obj)
+ assert res.exit_code == 1
+ assert res.exception is not None
+ assert isinstance(res.exception, SystemExit)
+ mock_run.assert_called_once()
+
+
+def test_auth_token_command_verifies_labels() -> None:
+ runner = CliRunner()
+ ctx_obj = {
+ 'config': {'targets': {'test': {'type': 'gmail', 'token_command': 'get-token'}}},
+ 'targets': {},
+ }
+ with (
+ patch('korgalore.google_creds.subprocess.run') as mock_run,
+ patch('korgalore.gmail_target.GmailTarget.list_labels') as mock_list_labels,
+ patch('korgalore.gmail_target.build'),
+ ):
+ mock_run.return_value = subprocess.CompletedProcess(
+ args=['get-token'], returncode=0, stdout='valid-token\n', stderr=''
+ )
+ res = runner.invoke(auth, ['test'], obj=ctx_obj)
+ assert res.exit_code == 0
+ mock_run.assert_called_once()
+ mock_list_labels.assert_called_once()
+
+
+def test_yank_token_command_failure_aborts_cleanly() -> None:
+ runner = CliRunner()
+ ctx_obj = {
+ 'config': {'targets': {'test': {'type': 'gmail', 'token_command': 'get-token'}}},
+ 'targets': {},
+ 'hide_bar': True,
+ }
+ with (
+ patch('korgalore.google_creds.subprocess.run') as mock_run,
+ patch('korgalore.cli.get_lore_node') as mock_node,
+ ):
+ mock_run.return_value = subprocess.CompletedProcess(
+ args=['get-token'], returncode=1, stdout='', stderr='command failed'
+ )
+ mock_node.return_value.get_mbox_by_msgid.return_value = b'From mboxrd@z Thu Jan 1 00:00:00 1970\n\n'
+ res = runner.invoke(yank, ['--target', 'test', '--thread', 'some-id'], obj=ctx_obj)
+ assert res.exit_code == 1
+ assert res.exception is not None
+ assert isinstance(res.exception, SystemExit)
+ mock_run.assert_called_once()
+
+
+def test_yank_single_message_command_oserror_aborts_cleanly() -> None:
+ runner = CliRunner()
+ ctx_obj = {
+ 'config': {'targets': {'test': {'type': 'gmail', 'token_command': '/nonexistent/get-token'}}},
+ 'targets': {},
+ 'hide_bar': True,
+ }
+ with (
+ patch('korgalore.google_creds.subprocess.run') as mock_run,
+ patch('korgalore.cli.get_lore_node') as mock_node,
+ ):
+ mock_run.side_effect = FileNotFoundError('No such file')
+ mock_node.return_value.get_message_by_msgid.return_value = b'From: a@b\nSubject: x\n\nBody\n'
+ res = runner.invoke(yank, ['--target', 'test', 'some-msgid'], obj=ctx_obj)
+ assert res.exit_code == 1
+ assert res.exception is not None
+ assert isinstance(res.exception, SystemExit)
+ mock_run.assert_called_once()
+
+
+def test_perform_yank_connection_failure_raises_remote_error() -> None:
+ ctx = make_ctx(
+ {
+ 'config': {'targets': {'test': {'type': 'gmail', 'token_command': 'get-token'}}},
+ 'targets': {},
+ 'hide_bar': True,
+ }
+ )
+ with patch('korgalore.google_creds.subprocess.run') as mock_run:
+ mock_run.return_value = subprocess.CompletedProcess(
+ args=['get-token'], returncode=1, stdout='', stderr='command failed'
+ )
+ with pytest.raises(RemoteError):
+ perform_yank(ctx, 'test', 'some-msgid')
diff --git a/tests/test_gmail_target.py b/tests/test_gmail_target.py
index 9e27f1c..3c5dfb2 100644
--- a/tests/test_gmail_target.py
+++ b/tests/test_gmail_target.py
@@ -10,7 +10,7 @@ from googleapiclient.errors import HttpError # type: ignore[import-untyped]
from korgalore import AuthenticationError, ConfigurationError, RemoteError
from korgalore.gmail_target import SCOPES, GmailTarget, make_gmail_credentials
-from korgalore.google_creds import OAuthFileCredentials
+from korgalore.google_creds import CommandCredentials, OAuthFileCredentials
def make_target() -> tuple[GmailTarget, MagicMock]:
@@ -300,18 +300,43 @@ class TestGmailTargetImportMessage:
class TestMakeGmailCredentials:
"""Tests for make_gmail_credentials configuration factory."""
- def test_no_credentials_raises(self) -> None:
- """Omitting credentials raises ConfigurationError."""
- with pytest.raises(ConfigurationError, match='No credentials file specified'):
+ def test_no_credentials_or_token_command_raises(self) -> None:
+ """Must specify either credentials or token_command."""
+ with pytest.raises(ConfigurationError, match='must specify either'):
make_gmail_credentials('test', {})
+ @pytest.mark.parametrize(
+ 'details',
+ [
+ {'credentials': '/tmp/creds.json', 'token_command': 'echo token'},
+ {'token': '/tmp/token.json', 'token_command': 'echo token'},
+ ],
+ ids=['credentials-and-command', 'token-and-command'],
+ )
+ def test_mutually_exclusive_token_command_options_raise(self, details: dict[str, object]) -> None:
+ """Specifying token_command alongside credentials or token raises ConfigurationError."""
+ with pytest.raises(ConfigurationError, match='cannot specify both'):
+ make_gmail_credentials('test', details)
+
+ def test_token_command_invalid_syntax_raises(self) -> None:
+ """Unclosed quotes in token_command raise ConfigurationError."""
+ with pytest.raises(ConfigurationError, match='syntax'):
+ make_gmail_credentials('test', {'token_command': 'unclosed "quote'})
+
+ def test_token_command_returns_command_credentials(self) -> None:
+ """Valid token_command builds CommandCredentials without running interactive OAuth."""
+ creds = make_gmail_credentials('test', {'token_command': 'get-token --profile work'})
+ assert isinstance(creds, CommandCredentials)
+ assert creds.command == ['get-token', '--profile', 'work']
+
@pytest.mark.parametrize(
('details', 'expected_match'),
[
({'credentials': 123}, "'credentials' must be a string"),
({'credentials': '/tmp/creds.json', 'token': 456}, "'token' must be a string"),
+ ({'token_command': 789}, "'token_command' must be a string"),
],
- ids=['non-str-credentials', 'non-str-token'],
+ ids=['non-str-credentials', 'non-str-token', 'non-str-token-command'],
)
def test_invalid_option_types_raise(self, details: dict[str, object], expected_match: str) -> None:
"""Non-string configuration values raise ConfigurationError."""
diff --git a/tests/test_google_creds.py b/tests/test_google_creds.py
index 4ac0a3d..f392a36 100644
--- a/tests/test_google_creds.py
+++ b/tests/test_google_creds.py
@@ -1,13 +1,19 @@
-"""Tests for Google OAuth2 credential providers."""
+"""Tests for Google OAuth2 and token_command credential providers."""
+import subprocess
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from google.auth.exceptions import RefreshError
-from korgalore import AuthenticationError, ConfigurationError
-from korgalore.google_creds import OAuthFileCredentials
+from korgalore import AuthenticationError, ConfigurationError, RemoteError
+from korgalore.google_creds import (
+ CommandCredentials,
+ OAuthFileCredentials,
+ parse_token_command,
+ run_token_command,
+)
TEST_SCOPES = ['https://example.invalid/scope.a', 'https://example.invalid/scope.b']
@@ -152,3 +158,105 @@ class TestOAuthFileCredentials:
provider = OAuthFileCredentials('test', str(creds), token, scopes=TEST_SCOPES, interactive=False)
assert provider.needs_auth
assert not provider.valid
+
+
+class TestTokenCommandHelpers:
+ """Tests for target-agnostic parse_token_command and run_token_command helpers."""
+
+ def test_parse_expands_env_and_home(self, tmp_path: Path) -> None:
+ """Splits quoted args and expands environment variables and tildes."""
+ with patch.dict('os.environ', {'HOME': str(tmp_path), 'HELPER_ARG': 'work account'}):
+ args = parse_token_command('~/bin/get-token --profile "$HELPER_ARG"', "target 'test'")
+
+ assert args == [f'{tmp_path}/bin/get-token', '--profile', 'work account']
+
+ @pytest.mark.parametrize(
+ ('value', 'expected_match'),
+ [
+ (123, 'must be a string'),
+ (' ', 'cannot be empty'),
+ ('unclosed "quote', 'syntax'),
+ ],
+ ids=['non-string', 'whitespace-only', 'bad-quoting'],
+ )
+ def test_parse_invalid_values_raise(self, value: object, expected_match: str) -> None:
+ """Invalid command strings raise ConfigurationError."""
+ with pytest.raises(ConfigurationError, match=expected_match):
+ parse_token_command(value, "target 'test'")
+
+ @patch('korgalore.google_creds.subprocess.run')
+ def test_run_returns_stripped_token(self, mock_run: MagicMock) -> None:
+ """Executes command without a shell and returns the stripped bearer token."""
+ mock_run.return_value = subprocess.CompletedProcess(
+ args=['get-token'], returncode=0, stdout=' valid-bearer-token\n', stderr=''
+ )
+
+ assert run_token_command(['get-token'], "target 'test'") == 'valid-bearer-token'
+ mock_run.assert_called_once_with(
+ ['get-token'],
+ stdin=subprocess.DEVNULL,
+ capture_output=True,
+ text=True,
+ encoding='utf-8',
+ errors='replace',
+ timeout=30,
+ check=False,
+ )
+
+ @patch('korgalore.google_creds.subprocess.run')
+ def test_run_nonzero_exit_raises_remote_error(self, mock_run: MagicMock) -> None:
+ """Non-zero exit raises RemoteError including the last non-empty stderr line."""
+ mock_run.return_value = subprocess.CompletedProcess(
+ args=['get-token'], returncode=1, stdout='', stderr='ignored line\nauth failed\n'
+ )
+
+ with pytest.raises(RemoteError, match='auth failed'):
+ run_token_command(['get-token'], "target 'test'")
+
+ @pytest.mark.parametrize(
+ 'stdout',
+ [' \n', 'multi word token\n'],
+ ids=['empty', 'embedded-whitespace'],
+ )
+ @patch('korgalore.google_creds.subprocess.run')
+ def test_run_invalid_output_raises_remote_error(self, mock_run: MagicMock, stdout: str) -> None:
+ """Empty or whitespace-containing stdout raises RemoteError."""
+ mock_run.return_value = subprocess.CompletedProcess(args=['get-token'], returncode=0, stdout=stdout, stderr='')
+
+ with pytest.raises(RemoteError, match='empty or invalid'):
+ run_token_command(['get-token'], "target 'test'")
+
+ @patch('korgalore.google_creds.subprocess.run')
+ def test_run_timeout_raises_remote_error(self, mock_run: MagicMock) -> None:
+ """Subprocess timeout raises RemoteError."""
+ mock_run.side_effect = subprocess.TimeoutExpired(cmd='get-token', timeout=30)
+
+ with pytest.raises(RemoteError, match='timed out after 30s'):
+ run_token_command(['get-token'], "target 'test'")
+
+ @patch('korgalore.google_creds.subprocess.run')
+ def test_run_oserror_raises_configuration_error(self, mock_run: MagicMock) -> None:
+ """Missing executable raises ConfigurationError."""
+ mock_run.side_effect = FileNotFoundError('No such file or directory')
+
+ with pytest.raises(ConfigurationError, match='Failed to execute'):
+ run_token_command(['nonexistent-cmd'], "target 'test'")
+
+
+class TestCommandCredentials:
+ """Tests for CommandCredentials."""
+
+ @patch('korgalore.google_creds.subprocess.run')
+ def test_refresh_sets_token_and_marks_valid(self, mock_run: MagicMock) -> None:
+ """CommandCredentials.refresh() invokes run_token_command and populates token."""
+ mock_run.return_value = subprocess.CompletedProcess(
+ args=['get-token'], returncode=0, stdout='valid-bearer-token\n', stderr=''
+ )
+
+ creds = CommandCredentials('test', ['get-token'])
+ assert not creds.valid
+
+ creds.refresh(None)
+
+ assert creds.valid
+ assert creds.token == 'valid-bearer-token'
--
2.56.0.542.g3ef89b7020-goog
^ permalink raw reply related [flat|nested] 4+ messages in thread