Raw Credential Brokering

google-auth
requests

import json
import time
from typing import Optional

from cachetools import TTLCache
from google.auth.transport.requests import Request as GoogleAuthRequest
from google.oauth2 import service_account

# Cache minted tokens: key = (service_name, scopes_tuple)
# TTL 50 min (tokens last 60 min, mint fresh 10 min before expiry)
_token_cache: TTLCache = TTLCache(maxsize=100, ttl=3000)

DEFAULT_SCOPES = ["https://www.googleapis.com/auth/cloud-platform"]


def is_gcp_service_account(credential_value: str) -> bool:
    """Check if a decrypted credential looks like a GCP SA JSON key."""
    try:
        data = json.loads(credential_value)
        return (
            isinstance(data, dict)
            and data.get("type") == "service_account"
            and "private_key" in data
            and "client_email" in data
        )
    except (json.JSONDecodeError, TypeError, ValueError):
        return False


def mint_access_token(
    credential_value: str,
    service_name: str,
    scopes: Optional[list[str]] = None,
    rid: str = "",
) -> dict:
    """Mint a short-lived GCP access token from a service account JSON key."""
    scopes = scopes or DEFAULT_SCOPES
    cache_key = (service_name, tuple(sorted(scopes)))

    # Check cache
    cached = _token_cache.get(cache_key)
    if cached:
        token, expiry_iso, email = cached
        return {
            "accessToken": token,
            "expiresAt": expiry_iso,
            "tokenType": "Bearer",
            "serviceAccount": email,
        }

    # Parse SA JSON and mint
    sa_info = json.loads(credential_value)
    email = sa_info.get("client_email", "unknown")

    credentials = service_account.Credentials.from_service_account_info(
        sa_info, scopes=scopes,
    )
    credentials.refresh(GoogleAuthRequest())

    access_token = credentials.token
    expiry = credentials.expiry  # datetime in UTC

    if expiry:
        expiry_iso = expiry.isoformat() + "Z"
    else:
        expiry_iso = time.strftime(
            "%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + 3600)
        )

    _token_cache[cache_key] = (access_token, expiry_iso, email)

    return {
        "accessToken": access_token,
        "expiresAt": expiry_iso,
        "tokenType": "Bearer",
        "serviceAccount": email,
    }

from gcp import is_gcp_service_account, mint_access_token

# ... inside the credential handler, after decryption ...

raw_access = token.get("accessToken", "")
if raw_access and is_gcp_service_account(raw_access):
    try:
        # Optional: parse scopes from query param
        scopes_param = request.query_params.get("scopes", "")
        scopes = [s.strip() for s in scopes_param.split(",") if s.strip()] or None

        minted = mint_access_token(raw_access, service, scopes=scopes, rid=rid)
        token = {
            "accessToken": minted["accessToken"],
            "tokenType": minted["tokenType"],
            "expiresAt": minted["expiresAt"],
            "serviceName": service,
            "serviceAccount": minted["serviceAccount"],
            "gcpMinted": True,
        }
    except Exception as gcp_err:
        return error_response(
            500, "gcp_mint_failed",
            f"Failed to mint GCP access token: {gcp_err}",
        )

from gcp import is_gcp_service_account, mint_access_token

# ... inside the proxy handler, after decrypting access_token ...

if is_gcp_service_account(access_token):
    try:
        minted = mint_access_token(access_token, service, rid=rid)
        access_token = minted["accessToken"]
    except Exception as gcp_err:
        return error_response(
            500, "gcp_mint_failed",
            f"Failed to mint GCP access token: {gcp_err}",
        )

# ... then inject access_token into upstream headers as before ...

def is_aws_credentials(value: str) -> bool:
    try:
        data = json.loads(value)
        return bool(
            data.get("aws_access_key_id")
            and data.get("aws_secret_access_key")
        )
    except (json.JSONDecodeError, TypeError):
        return False

import boto3

def mint_aws_session_token(
    credential_value: str, service_name: str, rid: str = ""
) -> dict:
    creds = json.loads(credential_value)
    sts = boto3.client(
        "sts",
        aws_access_key_id=creds["aws_access_key_id"],
        aws_secret_access_key=creds["aws_secret_access_key"],
    )
    session = sts.get_session_token(DurationSeconds=3600)["Credentials"]
    return {
        "accessToken": session["AccessKeyId"],
        "secretAccessKey": session["SecretAccessKey"],
        "sessionToken": session["SessionToken"],
        "expiresAt": session["Expiration"].isoformat(),
    }

if is_aws_credentials(raw_access):
    minted = mint_aws_session_token(raw_access, service, rid=rid)
    token = { ... }
elif is_gcp_service_account(raw_access):
    ...