feat(auth): add OAuth2 client flow + token store for GCP client onboarding

This commit is contained in:
chrischristiansen-glitch 2026-05-27 20:01:21 +02:00
parent 5c1693994e
commit 63f8bd289c
3 changed files with 168 additions and 0 deletions

1
auth/__init__.py Normal file
View File

@ -0,0 +1 @@
# auth package — OAuth2 client flow for OSVauco

91
auth/oauth_flow.py Normal file
View File

@ -0,0 +1,91 @@
"""
auth/oauth_flow.py Google OAuth2 flow for klient-onboarding.
Klienten trykker /auth/login?client_id=X
redirect til Google OAuth2 consent screen
Google redirecter til /auth/callback?code=...&state=...
token lagres i Secret Manager under klient-ID
klienten redirectes til /static/billing-dashboard.html
Scopes (read-only):
- bigquery.readonly
- cloud-billing.readonly
Krever env-vars:
OAUTH_CLIENT_ID fra GCP OAuth2 credentials
OAUTH_CLIENT_SECRET fra GCP OAuth2 credentials
OAUTH_REDIRECT_URI f.eks. https://osvauco-agent-....run.app/auth/callback
PROJECT_ID GCP project for Secret Manager
"""
import os
import json
import secrets
from google_auth_oauthlib.flow import Flow
SCOPES = [
"https://www.googleapis.com/auth/bigquery.readonly",
"https://www.googleapis.com/auth/cloud-billing.readonly",
"openid",
"https://www.googleapis.com/auth/userinfo.email",
]
_CLIENT_CONFIG = {
"web": {
"client_id": os.environ.get("OAUTH_CLIENT_ID", ""),
"client_secret": os.environ.get("OAUTH_CLIENT_SECRET", ""),
"redirect_uris": [os.environ.get("OAUTH_REDIRECT_URI", "http://localhost:8080/auth/callback")],
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"token_uri": "https://oauth2.googleapis.com/token",
}
}
# In-memory state store (nonce → client_id). For prod: bytt til Firestore/Redis.
_STATE_STORE: dict[str, str] = {}
def get_authorization_url(client_id: str) -> str:
"""
Genererer Google OAuth2 autoriseringsURL for gitt client_id.
Returnerer URL klienten skal redirectes til.
"""
flow = Flow.from_client_config(_CLIENT_CONFIG, scopes=SCOPES)
flow.redirect_uri = os.environ.get(
"OAUTH_REDIRECT_URI", "http://localhost:8080/auth/callback"
)
state = secrets.token_urlsafe(32)
_STATE_STORE[state] = client_id
auth_url, _ = flow.authorization_url(
access_type="offline",
include_granted_scopes="true",
state=state,
prompt="consent",
)
return auth_url
def exchange_code_for_token(code: str, state: str) -> tuple[str, dict]:
"""
Bytter OAuth2 code mot token.
Returnerer (client_id, token_dict).
Kaster ValueError hvis state er ukjent.
"""
client_id = _STATE_STORE.pop(state, None)
if client_id is None:
raise ValueError(f"Ukjent OAuth2 state: {state}")
flow = Flow.from_client_config(_CLIENT_CONFIG, scopes=SCOPES, state=state)
flow.redirect_uri = os.environ.get(
"OAUTH_REDIRECT_URI", "http://localhost:8080/auth/callback"
)
flow.fetch_token(code=code)
creds = flow.credentials
token_dict = {
"token": creds.token,
"refresh_token": creds.refresh_token,
"token_uri": creds.token_uri,
"client_id": creds.client_id,
"client_secret": creds.client_secret,
"scopes": list(creds.scopes or []),
}
return client_id, token_dict

76
auth/token_store.py Normal file
View File

@ -0,0 +1,76 @@
"""
auth/token_store.py Lagre og hente OAuth2-tokens per klient i Secret Manager.
Hvert token lagres som en JSON-streng under secret-navn:
osvauco-client-token-{client_id}
Bruk:
from auth.token_store import save_token, load_credentials
save_token("klient-abc", token_dict)
creds = load_credentials("klient-abc") # google.oauth2.credentials.Credentials
"""
import json
import os
from google.cloud import secretmanager
from google.oauth2.credentials import Credentials
PROJECT_ID = os.environ.get("PROJECT_ID", "propane-will-491900-m5")
def _secret_name(client_id: str) -> str:
return f"osvauco-client-token-{client_id}"
def save_token(client_id: str, token_dict: dict) -> None:
"""
Lagrer eller oppdaterer OAuth2-token for klient i Secret Manager.
Oppretter secret hvis den ikke finnes.
"""
client = secretmanager.SecretManagerServiceClient()
parent = f"projects/{PROJECT_ID}"
secret_id = _secret_name(client_id)
secret_path = f"{parent}/secrets/{secret_id}"
payload = json.dumps(token_dict).encode("utf-8")
# Opprett secret hvis den ikke finnes
try:
client.get_secret(name=secret_path)
except Exception:
client.create_secret(
request={
"parent": parent,
"secret_id": secret_id,
"secret": {"replication": {"automatic": {}}},
}
)
# Legg til ny versjon (ny verdi)
client.add_secret_version(
request={"parent": secret_path, "payload": {"data": payload}}
)
def load_credentials(client_id: str) -> Credentials:
"""
Henter siste token for klient fra Secret Manager.
Returnerer google.oauth2.credentials.Credentials klar til bruk.
Kaster KeyError hvis client_id ikke finnes.
"""
sm_client = secretmanager.SecretManagerServiceClient()
secret_path = f"projects/{PROJECT_ID}/secrets/{_secret_name(client_id)}/versions/latest"
try:
response = sm_client.access_secret_version(name=secret_path)
except Exception as exc:
raise KeyError(f"Ingen token funnet for klient '{client_id}': {exc}") from exc
token_dict = json.loads(response.payload.data.decode("utf-8"))
return Credentials(
token=token_dict.get("token"),
refresh_token=token_dict.get("refresh_token"),
token_uri=token_dict.get("token_uri", "https://oauth2.googleapis.com/token"),
client_id=token_dict.get("client_id"),
client_secret=token_dict.get("client_secret"),
scopes=token_dict.get("scopes"),
)