agilentics / boiler 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
"""Google sign-in: turning a Google-verified email into an account here.

The identity half of "Sign in with Google", kept transport-free like the rest of
app/core: the endpoint (app/endpoints/oauth.py) runs the OAuth redirect dance and
verifies the ID token; this module takes the resulting (google_sub, email, name)
and resolves it to a User, creating one - with an organisation - on first sign-in.

Three cases, in order:

  1. a user already linked to this Google subject -> return them
  2. a user with this email but no Google link    -> link and return. Safe
     because Google has verified the address, so this proves the same person
     rather than being a way to seize somebody's account.
  3. nobody yet                                    -> create the account and
     provision a tenant with them as owner. No password: an account that signs
     in only through a provider has no hash, which is what the nullable
     password_hash column exists to represent.
"""
import logging

from db.db_declarations import User

from . import auth as auth_core
from . import orgs as orgs_core

logger = logging.getLogger(__name__)


def find_or_create_google_user(session, google_sub: str, email: str,
                               name: str | None = None) -> User:
    email = (email or "").strip().lower()
    if not google_sub or "@" not in email:
        # The endpoint verifies the ID token before calling this, so a malformed
        # pair here is a programming error, not a user-facing one.
        raise ValueError("A Google sign-in needs a subject id and a verified email.")

    # 1. Already linked to this Google identity.
    user = session.query(User).filter(User.google_sub == google_sub).one_or_none()
    if user is not None:
        return user

    # 2. An existing account with this address - link it.
    user = session.query(User).filter(User.email == email).one_or_none()
    if user is not None:
        user.google_sub = google_sub
        session.commit()
        logger.info("Linked Google sign-in to existing account %s.", user.email)
        return user

    # 3. Brand new: the account, then a tenant with them as owner.
    user = auth_core.create_user(session, email, None, name=name)
    user.google_sub = google_sub

    # A Google account carries no organisation name, so the tenant is named
    # after the person. provision falls back to their address on its own.
    org = orgs_core.provision(session, user, name)
    session.commit()
    logger.info("Created account %s from a Google sign-in (new org %r).",
                user.email, org.slug)
    return user