JiriUhlir
278 lines
10 KiB
Python
278 lines
10 KiB
Python
"""Per-request credential extraction from X- headers (FastAPI dependencies).
|
|
|
|
The service is a STATELESS proxy: it stores no secrets. Every credential is
|
|
supplied per request as an X- header and used only to talk to the upstream API
|
|
(see AGENTS.md "Secrets v parametrech"). Declaring the headers as FastAPI
|
|
``Header`` parameters makes them appear per-operation in Swagger, including the
|
|
"Try it out" form.
|
|
|
|
All three Google services share the same OAuth mechanism (token wins over
|
|
service account); they differ only in the OAuth *scope* and the header prefix:
|
|
|
|
* Google Analytics -> ``X-GA-*`` (scope analytics.readonly)
|
|
* Search Console -> ``X-GSC-*`` (scope webmasters.readonly)
|
|
* Google Ads -> ``X-GAds-*`` (scope adwords) + developer token
|
|
|
|
Each provides ``*-Access-Token`` (ready Bearer token, takes precedence) and
|
|
``*-Credentials`` (base64 service-account JSON; the proxy mints a token). As an
|
|
alternative, the ready access token may also be supplied via the standard
|
|
``Authorization: Bearer <token>`` header (the service-specific ``*-Access-Token``
|
|
wins if both are present); this keeps the original header path intact while
|
|
offering the conventional bearer-token path.
|
|
|
|
Sklik uses ``X-Sklik-Token`` (the proxy calls client.loginByToken).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import json
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import Header
|
|
|
|
from . import config
|
|
from .errors import MissingCredentialsError
|
|
|
|
|
|
# --- Google (shared) ----------------------------------------------------------
|
|
@dataclass
|
|
class GoogleCredentials:
|
|
access_token: str | None
|
|
service_account_info: dict | None
|
|
quota_project: str | None
|
|
scope: str
|
|
|
|
|
|
@dataclass
|
|
class GoogleAdsCredentials:
|
|
google: GoogleCredentials
|
|
developer_token: str
|
|
login_customer_id: str | None
|
|
|
|
|
|
def _bearer_from_authorization(authorization: str | None) -> str | None:
|
|
"""Extract the token from a standard ``Authorization: Bearer <token>`` header.
|
|
|
|
Only the ``Bearer`` scheme is accepted; any other scheme (e.g. ``Basic``) is
|
|
ignored so the caller falls through to the other credential sources and gets
|
|
a clear "no credentials" error rather than a token that cannot work.
|
|
"""
|
|
if not authorization:
|
|
return None
|
|
parts = authorization.strip().split(None, 1)
|
|
if len(parts) == 2 and parts[0].lower() == "bearer":
|
|
return parts[1].strip() or None
|
|
return None
|
|
|
|
|
|
def _authorization_description(token_header: str) -> str:
|
|
return (
|
|
"Standard OAuth2 bearer token, sent as 'Authorization: Bearer <token>'. "
|
|
f"Alternative to {token_header} (which wins if both are present) and to "
|
|
"the service-account credentials header. Must carry the service's scope."
|
|
)
|
|
|
|
|
|
def _build_google_credentials(
|
|
access_token: str | None,
|
|
raw_credentials: str | None,
|
|
quota_project: str | None,
|
|
scope: str,
|
|
*,
|
|
token_header: str,
|
|
creds_header: str,
|
|
authorization: str | None = None,
|
|
) -> GoogleCredentials:
|
|
"""Parse a Google access token / base64 service-account JSON from headers.
|
|
|
|
The ready access token may arrive either in the service-specific
|
|
``*-Access-Token`` header (takes precedence, kept for backward
|
|
compatibility) or in a standard ``Authorization: Bearer <token>`` header.
|
|
"""
|
|
token = (access_token or "").strip() or None
|
|
if token is None:
|
|
token = _bearer_from_authorization(authorization)
|
|
|
|
service_account_info: dict | None = None
|
|
raw = (raw_credentials or "").strip()
|
|
if raw:
|
|
try:
|
|
decoded = base64.b64decode(raw, validate=True)
|
|
except (binascii.Error, ValueError) as exc:
|
|
raise MissingCredentialsError(
|
|
f"{creds_header} is not valid base64."
|
|
) from exc
|
|
try:
|
|
service_account_info = json.loads(decoded)
|
|
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
|
|
raise MissingCredentialsError(
|
|
f"{creds_header} does not decode to valid JSON."
|
|
) from exc
|
|
if not isinstance(service_account_info, dict):
|
|
raise MissingCredentialsError(
|
|
f"{creds_header} JSON must be a service-account object."
|
|
)
|
|
|
|
if not token and service_account_info is None:
|
|
raise MissingCredentialsError(
|
|
f"Provide either {token_header}, an 'Authorization: Bearer <token>' "
|
|
f"header, or {creds_header}."
|
|
)
|
|
|
|
return GoogleCredentials(
|
|
access_token=token,
|
|
service_account_info=service_account_info,
|
|
quota_project=(quota_project or "").strip() or None,
|
|
scope=scope,
|
|
)
|
|
|
|
|
|
# --- Google Analytics ---------------------------------------------------------
|
|
def get_ga_credentials(
|
|
x_ga_access_token: str | None = Header(
|
|
default=None,
|
|
alias="X-GA-Access-Token",
|
|
description="Ready OAuth2 access token used directly as a Bearer token. "
|
|
"Takes precedence over Authorization and X-GA-Credentials.",
|
|
),
|
|
authorization: str | None = Header(
|
|
default=None,
|
|
alias="Authorization",
|
|
description=_authorization_description("X-GA-Access-Token"),
|
|
),
|
|
x_ga_credentials: str | None = Header(
|
|
default=None,
|
|
alias="X-GA-Credentials",
|
|
description="Base64-encoded Google service-account JSON key. The proxy "
|
|
"mints a short-lived access token from it (scope analytics.readonly). "
|
|
"Used only if X-GA-Access-Token is absent.",
|
|
),
|
|
x_ga_quota_project: str | None = Header(
|
|
default=None,
|
|
alias="X-GA-Quota-Project",
|
|
description="Optional Google Cloud project id for quota/billing "
|
|
"(sets the x-goog-user-project header upstream).",
|
|
),
|
|
) -> GoogleCredentials:
|
|
return _build_google_credentials(
|
|
x_ga_access_token,
|
|
x_ga_credentials,
|
|
x_ga_quota_project,
|
|
config.GA_SCOPE,
|
|
token_header="X-GA-Access-Token",
|
|
creds_header="X-GA-Credentials",
|
|
authorization=authorization,
|
|
)
|
|
|
|
|
|
# --- Google Search Console ----------------------------------------------------
|
|
def get_gsc_credentials(
|
|
x_gsc_access_token: str | None = Header(
|
|
default=None,
|
|
alias="X-GSC-Access-Token",
|
|
description="Ready OAuth2 access token used directly as a Bearer token. "
|
|
"Takes precedence over Authorization and X-GSC-Credentials.",
|
|
),
|
|
authorization: str | None = Header(
|
|
default=None,
|
|
alias="Authorization",
|
|
description=_authorization_description("X-GSC-Access-Token"),
|
|
),
|
|
x_gsc_credentials: str | None = Header(
|
|
default=None,
|
|
alias="X-GSC-Credentials",
|
|
description="Base64-encoded Google service-account JSON key. The proxy "
|
|
"mints a short-lived access token from it (scope webmasters.readonly). "
|
|
"Used only if X-GSC-Access-Token is absent.",
|
|
),
|
|
x_gsc_quota_project: str | None = Header(
|
|
default=None,
|
|
alias="X-GSC-Quota-Project",
|
|
description="Optional Google Cloud project id for quota/billing.",
|
|
),
|
|
) -> GoogleCredentials:
|
|
return _build_google_credentials(
|
|
x_gsc_access_token,
|
|
x_gsc_credentials,
|
|
x_gsc_quota_project,
|
|
config.GSC_SCOPE,
|
|
token_header="X-GSC-Access-Token",
|
|
creds_header="X-GSC-Credentials",
|
|
authorization=authorization,
|
|
)
|
|
|
|
|
|
# --- Google Ads ---------------------------------------------------------------
|
|
def get_google_ads_credentials(
|
|
x_gads_developer_token: str | None = Header(
|
|
default=None,
|
|
alias="X-GAds-Developer-Token",
|
|
description="Google Ads API developer token (from a Google Ads manager "
|
|
"account). Required for every Google Ads call.",
|
|
),
|
|
x_gads_access_token: str | None = Header(
|
|
default=None,
|
|
alias="X-GAds-Access-Token",
|
|
description="Ready OAuth2 access token used directly as a Bearer token. "
|
|
"Takes precedence over Authorization and X-GAds-Credentials.",
|
|
),
|
|
authorization: str | None = Header(
|
|
default=None,
|
|
alias="Authorization",
|
|
description=_authorization_description("X-GAds-Access-Token"),
|
|
),
|
|
x_gads_credentials: str | None = Header(
|
|
default=None,
|
|
alias="X-GAds-Credentials",
|
|
description="Base64-encoded Google service-account JSON key (scope "
|
|
"adwords). Requires domain-wide delegation; usually an access token is "
|
|
"easier. Used only if X-GAds-Access-Token is absent.",
|
|
),
|
|
x_gads_login_customer_id: str | None = Header(
|
|
default=None,
|
|
alias="X-GAds-Login-Customer-Id",
|
|
description="Optional manager (MCC) customer id used as login-customer-id "
|
|
"header. Digits only, no dashes.",
|
|
),
|
|
x_gads_quota_project: str | None = Header(
|
|
default=None,
|
|
alias="X-GAds-Quota-Project",
|
|
description="Optional Google Cloud project id for quota/billing.",
|
|
),
|
|
) -> GoogleAdsCredentials:
|
|
developer_token = (x_gads_developer_token or "").strip()
|
|
if not developer_token:
|
|
raise MissingCredentialsError("X-GAds-Developer-Token header is required.")
|
|
|
|
google = _build_google_credentials(
|
|
x_gads_access_token,
|
|
x_gads_credentials,
|
|
x_gads_quota_project,
|
|
config.GOOGLE_ADS_SCOPE,
|
|
token_header="X-GAds-Access-Token",
|
|
creds_header="X-GAds-Credentials",
|
|
authorization=authorization,
|
|
)
|
|
login_customer_id = (x_gads_login_customer_id or "").strip().replace("-", "") or None
|
|
return GoogleAdsCredentials(
|
|
google=google,
|
|
developer_token=developer_token,
|
|
login_customer_id=login_customer_id,
|
|
)
|
|
|
|
|
|
# --- Sklik --------------------------------------------------------------------
|
|
def get_sklik_token(
|
|
x_sklik_token: str | None = Header(
|
|
default=None,
|
|
alias="X-Sklik-Token",
|
|
description="Sklik API token from Sklik account settings. The proxy uses "
|
|
"it to obtain a session via client.loginByToken.",
|
|
),
|
|
) -> str:
|
|
token = (x_sklik_token or "").strip()
|
|
if not token:
|
|
raise MissingCredentialsError("X-Sklik-Token header is required.")
|
|
return token
|