aeat.adapters.outbound.google._oauth_flow module

Google OAuth Desktop login flow for per-profile Google sessions.

Runs an operator-supplied adapters.outbound.google.OAuthClient through Google’s loopback IP + PKCE Desktop flow using google_auth_oauthlib.flow.InstalledAppFlow.run_local_server(port=0). The operating system picks an ephemeral loopback port and opens the consent screen in the operator’s default browser.

Two policy gates fire before any network IO happens:

  1. The caller must pass a profile identity resolved by adapters.outbound.google.resolve_active_profile().

  2. When core.config.SecretStoreBackend is configured as UNSECURED and that profile carries a real Spanish NIF / NIE / CIF, adapters.outbound.google._oauth_flow.check_unsecured_mode_safety() refuses with adapters.outbound.google.GoogleAuthUnsecuredModeRefusedError.

See also

adapters.outbound.google.run_login_flow() executes the login path, adapters.outbound.google._oauth_flow.credentials_to_records() produces adapters.outbound.google.OAuthToken and adapters.outbound.google.OAuthMetadata, and adapters.outbound.google.REQUIRED_SCOPES defines the consent surface the Google account must grant.

require_interactive_terminal()[source]

Refuse the consent flow when no controlling terminal can drive it.

The Desktop OAuth flow opens the consent screen in the operator’s browser and then blocks a loopback HTTP receiver until consent completes. In a non-interactive invocation (piped, redirected, cron, or another agent) stdin is not a TTY, no operator can complete the flow, and the receiver would block forever. This guard eliminates that silent-hang failure mode before adapters.outbound.google.run_login_flow() calls the local server.

Raises:

adapters.outbound.google.GoogleAuthNonInteractiveError – When sys.stdin is not attached to a terminal. The exception carries the interactive-terminal prerequisite as a suggestion.

Return type:

None

check_unsecured_mode_safety(profile, tax_id)[source]

Refuse the OAuth flow when unsecured mode meets a real NIF.

The guard mirrors the storage substrate’s NIF-canary rule: real taxpayer identifiers must not enter OAuth token setup while core.config.SecretStoreBackend is running in unsecured mode.

Parameters:
Raises:

adapters.outbound.google.GoogleAuthUnsecuredModeRefusedError – When aeat_secret_store_backend=unsecured and tax_id parses as a real Spanish tax identifier per adapters.persistence.storage.master_key.looks_like_real_tax_id().

Return type:

None

resolve_active_tax_id(profile_id)[source]

Return the identity.tax_id value for the profile UUID.

profile_id is the immutable profile identity returned by adapters.outbound.google.resolve_active_profile(). The resolver loads the profile bucket pointer through application.workflow.read_profile_bucket_by_id(), opens the canonical user-profile lifecycle service, and reads the tax-id fact used by adapters.outbound.google._oauth_flow.check_unsecured_mode_safety().

Return type:

str

Returns:

The stored identity.tax_id value, or an empty string when the profile record has no tax identifier.

Raises:

adapters.outbound.google.GoogleAuthProfileUnboundError – When the profile bucket manifest or canonical profile record cannot be resolved.

Parameters:

profile_id (str)

credentials_to_records(*, refresh_token, token_uri, account_email, granted_scopes, issued_at)[source]

Map OAuth credential fields into persisted Google session records.

The consent screen must grant every scope in adapters.outbound.google.REQUIRED_SCOPES. The returned adapters.outbound.google.OAuthToken carries the refresh credential and the returned adapters.outbound.google.OAuthMetadata carries the linked Google account, granted scope tuple, and issuance timestamps used by the session store.

Parameters:

  • refresh_token (str) – The refresh token returned by the consent screen.

  • token_uri (str) – The token endpoint URL mirrored from adapters.outbound.google.OAuthClient.

  • account_email (str) – The Google account that completed the consent.

  • granted_scopes (tuple[str, ...]) – Scopes the consent screen actually granted.

  • issued_at (datetime) – Timestamp the credential was first issued.

Return type:

tuple[OAuthToken, OAuthMetadata]

Returns:

A 2-tuple of (adapters.outbound.google.OAuthToken, adapters.outbound.google.OAuthMetadata) ready for adapters.persistence.storage.SecureObjectRepository persistence through adapters.outbound.google._session_store. Both records validate strict pydantic invariants; metadata refuses granted-scope tuples missing any adapters.outbound.google.REQUIRED_SCOPES member.

Raises:

adapters.outbound.google.GoogleAuthScopeInsufficientError – When granted_scopes omits any required scope. Re-raised separately from the pydantic ValidationError so the CLI can surface a concrete remediation hint.

run_login_flow(client, profile)[source]

Execute the loopback-IP + PKCE OAuth Desktop flow.

Always runs the real google_auth_oauthlib.flow.InstalledAppFlow.run_local_server(port=0) against accounts.google.com. The flow checks profile state with adapters.outbound.google._oauth_flow.resolve_active_tax_id(), applies adapters.outbound.google._oauth_flow.check_unsecured_mode_safety(), requires adapters.outbound.google._oauth_flow.require_interactive_terminal(), then maps the resulting credential fields through adapters.outbound.google._oauth_flow.credentials_to_records().

Parameters:
Return type:

tuple[OAuthToken, OAuthMetadata]

Returns:

A 2-tuple of (adapters.outbound.google.OAuthToken, adapters.outbound.google.OAuthMetadata) ready for persistence.

Raises:

adapters.outbound.google.GoogleAuthError – Any typed OAuth refusal with concrete remediation context.