| """ |
| Provider-agnostic sign-in logic shared by every OAuth provider (Google, Microsoft, ...). |
| Each provider's own auth_routes module (e.g. auth_routes.py for Google, ms_auth_routes.py for |
| Microsoft) handles the provider-specific redirect/token-exchange/id_token-verification dance, then |
| calls into this module for the identical "upsert user, accept invite, bootstrap/pick tenant" logic |
| that has nothing to do with which provider the user signed in with. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import os |
| from datetime import datetime |
|
|
| from fastapi import HTTPException, Request |
| from sqlalchemy.orm import Session |
|
|
| from .database import Invitation, Tenant, TenantMembership, User |
|
|
| from .oauth_utils import normalize_https_public_origin as _normalize_https_public_origin |
| from .oauth_utils import public_request_base as _public_request_base |
|
|
|
|
| def norm_email(s: str | None) -> str: |
| return (s or "").strip().lower() |
|
|
|
|
| def invite_token_hash(raw: str) -> str: |
| return hashlib.sha256(raw.strip().encode("utf-8")).hexdigest() |
|
|
|
|
| def post_login_url(request: Request) -> str: |
| fe = os.environ.get("FRONTEND_ORIGIN", "").strip() |
| if fe: |
| return _normalize_https_public_origin(fe) + "/" |
| return _public_request_base(request) + "/" |
|
|
|
|
| def upsert_user_by_provider( |
| db: Session, |
| *, |
| provider: str, |
| provider_sub: str, |
| email: str | None, |
| name: str | None, |
| picture: str | None, |
| ) -> User: |
| """Look up (or create) a User by their identity with the given provider. |
| |
| `provider` is "google" or "microsoft" — determines which column (`google_sub` / |
| `microsoft_sub`) is used to find/attach the identity. A user can end up with both set if they |
| later sign in with the other provider using the same email (each sign-in only ever touches its |
| own provider's column), but that linking is not automatic — two different Google/Microsoft |
| accounts create two separate User rows unless email happens to match at invite-acceptance time. |
| """ |
| if not provider_sub: |
| raise HTTPException(status_code=400, detail=f"Missing subject in {provider} identity token") |
| if provider == "microsoft": |
| u = db.query(User).filter(User.microsoft_sub == provider_sub).first() |
| else: |
| u = db.query(User).filter(User.google_sub == provider_sub).first() |
|
|
| email = email or "" |
| if u: |
| u.email = email or u.email |
| u.name = name or u.name |
| u.picture = picture or u.picture |
| if provider == "microsoft" and not u.microsoft_sub: |
| u.microsoft_sub = provider_sub |
| elif provider == "google" and not u.google_sub: |
| u.google_sub = provider_sub |
| else: |
| kwargs = {"email": email, "name": name, "picture": picture} |
| if provider == "microsoft": |
| kwargs["microsoft_sub"] = provider_sub |
| else: |
| kwargs["google_sub"] = provider_sub |
| u = User(**kwargs) |
| db.add(u) |
| db.flush() |
| return u |
|
|
|
|
| def try_accept_invite(db: Session, user: User, raw_token: str | None): |
| """ |
| Apply pending invitation. Returns tenant_id if membership was created or already existed via invite. |
| Raises ValueError('invite_email_mismatch') if email doesn't match. |
| Returns None if no token or invalid/expired invite (caller may bootstrap instead). |
| """ |
| if not raw_token: |
| return None |
| h = invite_token_hash(raw_token) |
| inv = db.query(Invitation).filter(Invitation.token_hash == h).first() |
| if not inv or inv.accepted_at is not None: |
| return None |
| if inv.expires_at < datetime.utcnow(): |
| return None |
| if norm_email(user.email) != norm_email(inv.email): |
| raise ValueError("invite_email_mismatch") |
| existing = ( |
| db.query(TenantMembership) |
| .filter( |
| TenantMembership.user_id == user.id, |
| TenantMembership.tenant_id == inv.tenant_id, |
| ) |
| .first() |
| ) |
| if not existing: |
| db.add( |
| TenantMembership( |
| user_id=user.id, |
| tenant_id=inv.tenant_id, |
| role=inv.role or "member", |
| team_id=inv.team_id, |
| ) |
| ) |
| inv.accepted_at = datetime.utcnow() |
| return inv.tenant_id |
|
|
|
|
| def bootstrap_membership_if_needed(db: Session, user: User) -> None: |
| if db.query(TenantMembership).filter(TenantMembership.user_id == user.id).count() > 0: |
| return |
| total_m = db.query(TenantMembership).count() |
| if total_m == 0: |
| t = db.query(Tenant).order_by(Tenant.id).first() |
| if not t: |
| t = Tenant(name="My workspace") |
| db.add(t) |
| db.flush() |
| db.add(TenantMembership(user_id=user.id, tenant_id=t.id, role="admin")) |
| else: |
| label = (user.email or "user").split("@")[0] |
| t = Tenant(name=f"{label}'s workspace") |
| db.add(t) |
| db.flush() |
| db.add(TenantMembership(user_id=user.id, tenant_id=t.id, role="admin")) |
|
|
|
|
| def pick_current_tenant_id(db: Session, user: User, invite_tid: int | None) -> int | None: |
| if invite_tid is not None: |
| m = ( |
| db.query(TenantMembership) |
| .filter( |
| TenantMembership.user_id == user.id, |
| TenantMembership.tenant_id == invite_tid, |
| ) |
| .first() |
| ) |
| if m: |
| return int(invite_tid) |
| m = ( |
| db.query(TenantMembership) |
| .filter(TenantMembership.user_id == user.id) |
| .order_by(TenantMembership.id) |
| .first() |
| ) |
| return int(m.tenant_id) if m else None |
|
|
