workspace_onboarding_ux.py

"""
Overengineered web form to facilitate onboarding users to Google Workspace
"""

import logging
import os
from base64 import b64encode
from email.errors import HeaderParseError
from email.headerregistry import Address
from hashlib import file_digest
from json import loads
from re import fullmatch
from typing import Any, Dict, Union
from urllib.parse import urlparse, urlunparse
from uuid import UUID, uuid4

from authlib.integrations.flask_client import OAuth
from authlib.integrations.requests_client import OAuth2Session

from celery import Celery, Task, shared_task

from flask import Flask, render_template, request, session
from flask.helpers import get_debug_flag, redirect, url_for

from flask_caching import Cache

from flask_minify import Minify  # type: ignore

from google.oauth2 import service_account

from googleapiclient.discovery import build  # type: ignore
from googleapiclient.errors import HttpError  # type: ignore

from hubspot import HubSpot  # type: ignore
from hubspot.settings.users.exceptions import NotFoundException  # type: ignore

from ldap3 import Connection, Server

from requests import post

import sentry_sdk
from sentry_sdk import set_user
from sentry_sdk.integrations.flask import FlaskIntegration
from sentry_sdk.integrations.pure_eval import PureEvalIntegration

from slack_sdk import WebClient, WebhookClient
from slack_sdk.errors import SlackApiError
from slack_sdk.models.blocks import ActionsBlock, ButtonElement, ConfirmObject, SectionBlock
from slack_sdk.signature import SignatureVerifier

from werkzeug.exceptions import BadRequest, Conflict, Unauthorized

USER_AGENT = (
    "WorkspaceOnboarding/"
    + os.environ.get("NOMAD_TASK_NAME", "local")
    + "/"
    + os.environ.get("NOMAD_SHORT_ALLOC_ID", "local")
)

NAME_PATTERN = r"^[a-zA-Z'.\- ]+$"


def traces_sampler(sampling_context: Dict[str, Dict[str, str]]) -> bool:
    """
    Ignore ping events, sample all other events
    """
    try:
        request_uri = sampling_context["wsgi_environ"]["REQUEST_URI"]
    except KeyError:
        return False

    return request_uri != "/ping"


def init_celery(flask: Flask) -> Celery:
    """
    Initialize Celery
    """

    class FlaskTask(Task):  # type: ignore  # pylint: disable=abstract-method
        """
        Extend default Task class to have Flask context available

        https://flask.palletsprojects.com/en/stable/patterns/celery/
        """

        def __call__(self, *args, **kwargs):  # type: ignore
            with flask.app_context():
                return self.run(*args, **kwargs)

    new_celery_app = Celery("workspace_onboarding_ux", task_cls=FlaskTask)
    new_celery_app.config_from_object(flask.config, namespace="CELERY")
    new_celery_app.set_default()
    flask.extensions["celery"] = new_celery_app
    return new_celery_app  # type: ignore


sentry_sdk.init(
    debug=get_debug_flag(),
    integrations=[
        FlaskIntegration(),
        PureEvalIntegration(),
    ],
    traces_sampler=traces_sampler,
    attach_stacktrace=True,
    max_request_body_size="always",
    in_app_include=[
        "workspace_onboarding_ux",
    ],
    profiles_sample_rate=1.0,
)

app = Flask(__name__)

if not get_debug_flag():
    Minify(app=app, caching_limit=0, go=False)

app.config.from_prefixed_env()

keycloak_server = urlunparse(
    (
        urlparse(app.config["KEYCLOAK_METADATA_URL"]).scheme,
        urlparse(app.config["KEYCLOAK_METADATA_URL"]).netloc,
        "",
        "",
        "",
        "",
    )
)

celery_app = init_celery(app)

oauth = OAuth(app)  # type: ignore
oauth.register(  # type: ignore
    name="keycloak",
    server_metadata_url=app.config["KEYCLOAK_METADATA_URL"],
    client_kwargs={"scope": "openid email profile"},
)

keycloak = OAuth2Session(
    client_id=app.config["KEYCLOAK_ADMIN_CLIENT_ID"],
    client_secret=app.config["KEYCLOAK_ADMIN_CLIENT_SECRET"],
    token_endpoint=urlunparse(
        (
            urlparse(app.config["KEYCLOAK_METADATA_URL"]).scheme,
            urlparse(app.config["KEYCLOAK_METADATA_URL"]).netloc,
            "/realms/master/protocol/openid-connect/token",
            "",
            "",
            "",
        )
    ),
    leeway=5,
)
keycloak.headers["User-Agent"] = USER_AGENT  # type: ignore[attr-defined]
keycloak.fetch_token()

apiary = OAuth2Session(
    client_id=app.config["APIARY_CLIENT_ID"],
    client_secret=app.config["APIARY_CLIENT_SECRET"],
    token_endpoint=app.config["APIARY_URL"] + "/oauth/token",
    leeway=300,  # Discard tokens 5 minutes before expiration
)
apiary.headers["User-Agent"] = USER_AGENT  # type: ignore[attr-defined]
apiary.fetch_token()

google_workspace = build(
    serviceName="admin",
    version="directory_v1",
    credentials=service_account.Credentials.from_service_account_info(  # type: ignore
        info=app.config["GOOGLE_SERVICE_ACCOUNT_CREDENTIALS"],
        scopes=["https://www.googleapis.com/auth/admin.directory.user"],
        subject=app.config["GOOGLE_SUBJECT"],
    ),
).users()

cache = Cache(app)

logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
req_log = logging.getLogger("urllib3")
req_log.setLevel(logging.DEBUG)
req_log.propagate = True


def generate_subresource_integrity_hash(file: str) -> str:
    """
    Calculate the subresource integrity hash for a given file
    """
    with open(file[1:], "rb") as f:
        d = file_digest(f, "sha512")

    return "sha512-" + b64encode(d.digest()).decode("utf-8")


app.jinja_env.globals["calculate_integrity"] = generate_subresource_integrity_hash


@cache.memoize(timeout=0, cache_none=True)
def get_slack_user_id_by_email(email: str) -> Union[str, None]:
    """
    Wrapper for the users.lookupByEmail function to memoize responses
    """
    slack = WebClient(token=app.config["SLACK_API_TOKEN"])

    try:
        slack_response = slack.users_lookupByEmail(email=email)

        if slack_response.data["ok"]:  # type: ignore
            return slack_response.data["user"]["id"]  # type: ignore
    except SlackApiError:
        # this exception is thrown if there is no user with this email (among other possibilities)
        pass

    return None


@cache.memoize(timeout=0, cache_none=True)
def get_slack_user_id(  # pylint: disable=too-many-return-statements,too-many-branches
    keycloak_user_id: str,
) -> Union[str, None]:
    """
    Get the Slack user ID for a given Keycloak user
    """
    get_keycloak_user_response = keycloak.get(  # type: ignore
        url=keycloak_server
        + "/admin/realms/"
        + app.config["KEYCLOAK_REALM"]
        + "/users/"
        + keycloak_user_id,
        timeout=(5, 5),
    )
    get_keycloak_user_response.raise_for_status()

    keycloak_user = get_keycloak_user_response.json()

    if (
        "attributes" in keycloak_user
        and keycloak_user["attributes"] is not None
        and "googleWorkspaceAccount" in keycloak_user["attributes"]
        and keycloak_user["attributes"]["googleWorkspaceAccount"] is not None
        and len(keycloak_user["attributes"]["googleWorkspaceAccount"]) > 0
    ):
        slack_user_id = get_slack_user_id_by_email(
            keycloak_user["attributes"]["googleWorkspaceAccount"][0]
        )

        if slack_user_id is not None:
            return slack_user_id  # type: ignore

    if "email" in keycloak_user and keycloak_user["email"] is not None:
        slack_user_id = get_slack_user_id_by_email(keycloak_user["email"])

        if slack_user_id is not None:
            return slack_user_id  # type: ignore

    if "username" in keycloak_user and keycloak_user["username"] is not None:
        slack_user_id = get_slack_user_id_by_email(keycloak_user["username"] + "@gatech.edu")

        if slack_user_id is not None:
            return slack_user_id  # type: ignore

    if "username" in keycloak_user and keycloak_user["username"] is not None:
        apiary_user_response = apiary.get(  # type: ignore
            url=app.config["APIARY_URL"] + "/api/v1/users/" + keycloak_user["username"],
            headers={"Accept": "application/json"},
            timeout=(5, 5),
        )

        if apiary_user_response.status_code == 200:
            apiary_user = apiary_user_response.json()["user"]

            if "gt_email" in apiary_user and apiary_user["gt_email"] is not None:
                slack_user_id = get_slack_user_id_by_email(apiary_user["gt_email"])

                if slack_user_id is not None:
                    return slack_user_id  # type: ignore

            if "gmail_address" in apiary_user and apiary_user["gmail_address"] is not None:
                slack_user_id = get_slack_user_id_by_email(apiary_user["gmail_address"])

                if slack_user_id is not None:
                    return slack_user_id  # type: ignore

            if "clickup_email" in apiary_user and apiary_user["clickup_email"] is not None:
                slack_user_id = get_slack_user_id_by_email(apiary_user["clickup_email"])

                if slack_user_id is not None:
                    return slack_user_id  # type: ignore

    with sentry_sdk.start_span(op="ldap.connect"):
        ldap = Connection(
            Server("whitepages.gatech.edu", connect_timeout=1),
            auto_bind=True,
            raise_exceptions=True,
            receive_timeout=1,
        )
    with sentry_sdk.start_span(op="ldap.search"):
        result = ldap.search(
            search_base="dc=whitepages,dc=gatech,dc=edu",
            search_filter="(uid=" + keycloak_user["username"] + ")",
            attributes=["mail"],
        )

    if result is True:
        for entry in ldap.entries:
            if "mail" in entry and entry["mail"] is not None and entry["mail"].value is not None:
                slack_user_id = get_slack_user_id_by_email(entry["mail"].value)

                if slack_user_id is not None:
                    return slack_user_id  # type: ignore

    return None


@cache.cached(timeout=0, key_prefix="slack_team_id")
def get_slack_team_id() -> str:
    """
    Get the team ID for the bot user, used for generating deep links

    https://docs.slack.dev/interactivity/deep-linking#open_a_channel
    """
    slack = WebClient(token=app.config["SLACK_API_TOKEN"])

    slack_response = slack.team_info()

    return slack_response["team"]["id"]  # type: ignore


@cache.memoize(timeout=0)
def get_slack_channel_name(channel_id: str) -> str:
    """
    Get the channel name for the given channel ID
    """
    slack = WebClient(token=app.config["SLACK_API_TOKEN"])

    slack_response = slack.conversations_info(channel=channel_id)

    return slack_response["channel"]["name"]  # type: ignore


@shared_task
def remove_eligible_role(keycloak_user_id: str) -> None:
    """
    Remove the eligible role from this user in Keycloak, after they are provisioned
    """
    remove_eligible_role_response = keycloak.delete(  # type: ignore
        url=keycloak_server
        + "/admin/realms/"
        + app.config["KEYCLOAK_REALM"]
        + "/users/"
        + keycloak_user_id
        + "/role-mappings/clients/"
        + app.config["KEYCLOAK_CLIENT_UUID"],
        timeout=(5, 5),
        json=[{"id": app.config["KEYCLOAK_CLIENT_ROLE_ELIGIBLE"], "name": "eligible"}],
    )
    remove_eligible_role_response.raise_for_status()


@shared_task(max_retries=0)
def import_user_to_org_chart(workspace_user_id: str) -> None:
    """
    Notify OrgChart after a user is added to Google Workspace
    """
    org_chart_response = post(
        url=app.config["ORG_CHART_NOTIFY_URL"],
        headers={
            "Accept": "application/json",
            "Authorization": "Token " + app.config["ORG_CHART_TOKEN"],
            "User-Agent": USER_AGENT,
        },
        timeout=(5, 5),
        json={"google_workspace_user_id": workspace_user_id},
    )
    org_chart_response.raise_for_status()


@shared_task
def notify_slack_ineligible(keycloak_user_id: str) -> None:
    """
    Send a Slack notification to the central notifications channel when an ineligible user loads
    the form
    """
    if cache.get("slack_ineligible_message_" + keycloak_user_id) is not None:
        return

    get_keycloak_user_response = keycloak.get(  # type: ignore
        url=keycloak_server
        + "/admin/realms/"
        + app.config["KEYCLOAK_REALM"]
        + "/users/"
        + keycloak_user_id,
        timeout=(5, 5),
    )
    get_keycloak_user_response.raise_for_status()

    view_in_keycloak_button = ButtonElement(
        text="View in Keycloak",
        action_id="view_in_keycloak",
        url=urlunparse(
            (
                urlparse(app.config["KEYCLOAK_METADATA_URL"]).scheme,
                urlparse(app.config["KEYCLOAK_METADATA_URL"]).hostname,
                "/admin/master/console/",
                "",
                "",
                "/"
                + app.config["KEYCLOAK_REALM"]
                + "/users/"
                + str(keycloak_user_id)
                + "/settings",
            )
        ),
    )

    apiary_user_response = apiary.get(  # type: ignore
        url=app.config["APIARY_URL"]
        + "/api/v1/users/"
        + get_keycloak_user_response.json()["username"],
        headers={"Accept": "application/json"},
        timeout=(5, 5),
    )

    if apiary_user_response.status_code == 200:
        personal_pronoun_is = "they are"

        if (
            "gender" in apiary_user_response.json()["user"]
            and apiary_user_response.json()["user"]["gender"] is not None
        ):
            if str.lower(apiary_user_response.json()["user"]["gender"]) == "male":
                personal_pronoun_is = "he is"
            elif str.lower(apiary_user_response.json()["user"]["gender"]) == "female":
                personal_pronoun_is = "she is"

        actions = ActionsBlock(
            elements=[
                ButtonElement(
                    text="View in Apiary",
                    action_id="view_in_apiary",
                    url=app.config["APIARY_URL"]
                    + "/nova/resources/users/"
                    + str(apiary_user_response.json()["user"]["id"]),
                ),
                view_in_keycloak_button,
                ButtonElement(
                    text="Grant Eligibility in Keycloak",
                    action_id="grant_eligibility_in_keycloak",
                    value=keycloak_user_id,
                    style="primary",
                    confirm=ConfirmObject(
                        title="Grant Eligibility in Keycloak",
                        text="Are you sure you want to grant "
                        + get_keycloak_user_response.json()["firstName"]
                        + " eligibility for a Google Workspace account in Keycloak? If "
                        + personal_pronoun_is
                        + " in a leadership role, you should assign a role within Apiary instead.",  # noqa
                        confirm="Grant Eligibility",
                        deny="Cancel",
                    ),
                ),
            ]
        )
    elif apiary_user_response.status_code == 404:
        actions = ActionsBlock(
            elements=[
                view_in_keycloak_button,
            ]
        )
    else:
        actions = ActionsBlock(elements=[])
        apiary_user_response.raise_for_status()

    slack_user_id = get_slack_user_id(keycloak_user_id=keycloak_user_id)

    user_name = (
        get_keycloak_user_response.json()["firstName"]
        + " "
        + get_keycloak_user_response.json()["lastName"]
    )

    if slack_user_id is None:
        user_mention = user_name
    else:
        user_mention = f"<@{slack_user_id}>"

    slack = WebClient(token=app.config["SLACK_API_TOKEN"])

    slack_response = slack.chat_postMessage(
        channel=app.config["SLACK_NOTIFY_CHANNEL"],
        text=user_name
        + " logged in to the Google Workspace onboarding form, but isn't eligible for a Google Workspace account.",  # noqa
        blocks=[
            SectionBlock(
                text=user_mention
                + " logged in to the Google Workspace onboarding form, but isn't eligible for a Google Workspace account."  # noqa
            ),
            actions,
        ],
    )

    cache.set("slack_ineligible_message_" + keycloak_user_id, slack_response["ts"])


@shared_task
def notify_slack_account_created(keycloak_user_id: str) -> None:
    """
    Send Slack notifications to the central notifications channel when a new user is added
    to Google Workspace
    """
    keycloak_user_response = keycloak.get(  # type: ignore
        url=keycloak_server
        + "/admin/realms/"
        + app.config["KEYCLOAK_REALM"]
        + "/users/"
        + keycloak_user_id,
        timeout=(5, 5),
    )
    keycloak_user_response.raise_for_status()

    user_json = keycloak_user_response.json()

    new_user_slack_user_id = get_slack_user_id(keycloak_user_id=keycloak_user_id)

    slack = WebClient(token=app.config["SLACK_API_TOKEN"])

    new_user_name = user_json["firstName"] + " " + user_json["lastName"]

    if new_user_slack_user_id is None:
        new_user_mention = new_user_name
    else:
        new_user_mention = f"<@{new_user_slack_user_id}>"

    slack.chat_postMessage(
        channel=app.config["SLACK_NOTIFY_CHANNEL"],
        thread_ts=cache.get("slack_ineligible_message_" + keycloak_user_id),
        reply_broadcast=True,
        text=new_user_name + " joined Google Workspace!",
        blocks=[
            SectionBlock(
                text=new_user_mention + " joined Google Workspace!",
            ),
            ActionsBlock(
                elements=[
                    ButtonElement(
                        text="View in Google Workspace",
                        action_id="view_in_workspace",
                        url="https://www.google.com/a/robojackets.org/ServiceLogin?continue=https://admin.google.com/ac/search?query="  # noqa
                        + user_json["attributes"]["googleWorkspaceAccount"][0]
                        + "&tab=USERS",
                    )
                ]
            ),
        ],
    )


@shared_task(
    bind=True,
    max_retries=10,
    default_retry_delay=10,
    retry_backoff=True,
    retry_jitter=True,
    retry_backoff_max=60,
)
def invite_user_to_hubspot(self: Task, google_workspace_user_id: str) -> None:  # type: ignore
    """
    Invite a Google Workspace user to HubSpot
    """
    try:
        workspace_user = google_workspace.get(userKey=google_workspace_user_id).execute()
    except HttpError as e:
        if e.status_code == 404:
            raise self.retry(exc=e) from e

        raise e

    if not workspace_user["isMailboxSetup"]:
        raise self.retry(exc=Exception("Mailbox is not ready yet"))

    if not workspace_user["agreedToTerms"]:
        raise self.retry(exc=Exception("User has not agreed to terms yet"))

    hubspot = HubSpot(access_token=app.config["HUBSPOT_ACCESS_TOKEN"])

    try:
        hubspot.settings.users.users_api.get_by_id(
            user_id=workspace_user["primaryEmail"], id_property="EMAIL"
        )
    except NotFoundException:
        hubspot.settings.users.users_api.create(
            user_provision_request={
                "firstName": workspace_user["name"]["givenName"],
                "lastName": workspace_user["name"]["familyName"],
                "email": workspace_user["primaryEmail"],
                "sendWelcomeEmail": True,
            }
        )


def validate_name(which: str, value: str) -> str:
    """
    Validate a first or last name; return the stripped value or raise BadRequest
    """
    stripped = value.strip()

    if stripped == "":
        raise BadRequest("Please enter your " + which + " name")

    if len(stripped) < 2:
        raise BadRequest("Your " + which + " name must be at least 2 characters")

    if len(stripped) > 60:
        raise BadRequest("Your " + which + " name may be a maximum of 60 characters")

    if fullmatch(NAME_PATTERN, stripped) is None:
        raise BadRequest(
            "Your "
            + which
            + " name may only contain letters, spaces, dashes, apostrophes, and periods"
        )

    return stripped


def validate_email_address(value: str) -> str:
    """
    Validate a requested @robojackets.org address; return stripped lowercase or raise BadRequest
    """
    stripped = value.strip().lower()

    try:
        address = Address(addr_spec=stripped)
    except (ValueError, IndexError, TypeError, HeaderParseError) as exc:
        raise BadRequest("Please enter a valid email address") from exc

    if address.domain != "robojackets.org":
        raise BadRequest("Your email address must end in @robojackets.org")

    local_parts = address.username.split(".")

    if len(local_parts) != 2:
        raise BadRequest(
            "Your email address should include your first and last name separated by a period"
        )

    if len(local_parts[0]) < 2:
        raise BadRequest("Your first name must be at least 2 characters")

    if len(local_parts[1]) < 2:
        raise BadRequest("Your last name must be at least 2 characters")

    if len(address.username) > 60:
        raise BadRequest(
            "Your email address may be a maximum of 60 characters followed by @robojackets.org"
        )

    if fullmatch(NAME_PATTERN, address.username) is None:
        raise BadRequest("Your email address may only contain letters, dashes, and periods")

    return stripped


@cache.memoize(timeout=0)
def is_email_available(email: str) -> bool:
    """
    Return True if the email is not already used in Keycloak or Google Workspace
    """
    search_keycloak_user_response = keycloak.get(  # type: ignore
        url=keycloak_server + "/admin/realms/" + app.config["KEYCLOAK_REALM"] + "/users",
        params={
            "q": "googleWorkspaceAccount:" + email,
        },
        timeout=(5, 5),
    )
    search_keycloak_user_response.raise_for_status()

    if len(search_keycloak_user_response.json()) > 0:
        return False

    try:
        google_workspace.get(userKey=email).execute()

        return False
    except HttpError as e:
        if e.status_code != 404:
            raise e

    return True


@app.get("/")
def index() -> Any:
    """
    Generates the main form, or messaging if the user shouldn't fill it out
    """
    if "user_state" not in session:
        return oauth.keycloak.authorize_redirect(url_for("login", _external=True))

    set_user(
        {
            "id": session["sub"],
            "ip_address": request.remote_addr,
        }
    )

    if session["user_state"] == "provisioned":
        return render_template(
            "provisioned.html",
            workspace_account=session["email_address"],
            slack_team_id=get_slack_team_id(),
            slack_support_channel_id=app.config["SLACK_SUPPORT_CHANNEL"],
            slack_support_channel_name=get_slack_channel_name(app.config["SLACK_SUPPORT_CHANNEL"]),
        )

    keycloak_user_response = keycloak.get(  # type: ignore
        url=keycloak_server
        + "/admin/realms/"
        + app.config["KEYCLOAK_REALM"]
        + "/users/"
        + session["sub"],
        timeout=(5, 5),
    )
    keycloak_user_response.raise_for_status()

    user_json = keycloak_user_response.json()
    attributes = user_json["attributes"] if "attributes" in user_json else {}
    google_workspace_account = (
        attributes["googleWorkspaceAccount"][0]
        if "googleWorkspaceAccount" in attributes and len(attributes["googleWorkspaceAccount"]) > 0
        else None
    )

    if google_workspace_account is not None:
        workspace_user = google_workspace.get(userKey=google_workspace_account).execute()

        session["user_state"] = "provisioned"
        session["email_address"] = workspace_user["primaryEmail"]

        invite_user_to_hubspot.delay(workspace_user["id"])

        return render_template(
            "provisioned.html",
            workspace_account=session["email_address"],
            slack_team_id=get_slack_team_id(),
            slack_support_channel_id=app.config["SLACK_SUPPORT_CHANNEL"],
            slack_support_channel_name=get_slack_channel_name(app.config["SLACK_SUPPORT_CHANNEL"]),
        )

    if session["user_state"] == "ineligible":
        session.clear()
        return (
            render_template(
                "ineligible.html",
                slack_team_id=get_slack_team_id(),
                slack_support_channel_id=app.config["SLACK_SUPPORT_CHANNEL"],
                slack_support_channel_name=get_slack_channel_name(
                    app.config["SLACK_SUPPORT_CHANNEL"]
                ),
            ),
            424,
        )

    return render_template(
        "form.html",
        elm_model={
            "firstName": session["first_name"],
            "lastName": session["last_name"],
            "emailAddress": session["email_address"],
        },
    )


@app.get("/login")
def login() -> Any:  # pylint: disable=too-many-branches
    """
    Handles the return from Keycloak and collects default values for the form
    """
    token = oauth.keycloak.authorize_access_token()

    userinfo = token["userinfo"]

    session["sub"] = userinfo["sub"]

    set_user(
        {
            "id": session["sub"],
            "ip_address": request.remote_addr,
        }
    )

    if "googleWorkspaceAccount" in userinfo and userinfo["googleWorkspaceAccount"] is not None:
        workspace_user = google_workspace.get(userKey=userinfo["googleWorkspaceAccount"]).execute()

        session["user_state"] = "provisioned"
        session["email_address"] = workspace_user["primaryEmail"]

        invite_user_to_hubspot.delay(workspace_user["id"])

        return redirect(url_for("index"))

    session["first_name"] = (
        userinfo["given_name"]
        if "given_name" in userinfo and userinfo["given_name"] != "Confidential"
        else ""
    )
    session["last_name"] = (
        userinfo["family_name"]
        if "family_name" in userinfo and userinfo["family_name"] != "Confidential"
        else ""
    )
    session["email_address"] = (
        session["first_name"] + "." + session["last_name"] + "@robojackets.org"
    ).lower()

    if "roles" in userinfo and "eligible" in userinfo["roles"]:
        session["user_state"] = "eligible"
    else:
        session["user_state"] = "ineligible"

    apiary_user_response = apiary.get(  # type: ignore
        url=app.config["APIARY_URL"] + "/api/v1/users/" + userinfo["preferred_username"],
        headers={
            "Accept": "application/json",
        },
        params={"include": "roles,teams"},
        timeout=(5, 5),
    )

    if apiary_user_response.status_code == 200:
        apiary_user = apiary_user_response.json()["user"]

        role_check = False

        if "roles" in apiary_user and apiary_user["roles"] is not None:
            for role in apiary_user["roles"]:
                if role["name"] != "member" and role["name"] != "non-member":
                    role_check = True

        if (
            apiary_user["is_active"]
            and apiary_user["is_access_active"]
            and apiary_user["signed_latest_agreement"]
            and len(apiary_user["teams"]) > 0
            and role_check
        ):
            session["user_state"] = "eligible"

    elif apiary_user_response.status_code == 404:
        session["user_state"] = "ineligible"

    else:
        apiary_user_response.raise_for_status()

    if session["user_state"] == "ineligible":
        notify_slack_ineligible.delay(userinfo["sub"])

    return redirect(url_for("index"))


@app.post("/check-availability")
def check_availability() -> Any:
    """
    Check if a given email address is available for use
    """
    if "user_state" not in session:
        raise Unauthorized("Not logged in")

    if session["user_state"] != "eligible":
        raise Unauthorized("Not eligible")

    set_user(
        {
            "id": session["sub"],
            "ip_address": request.remote_addr,
        }
    )

    body = request.get_json(silent=True)

    if not isinstance(body, dict) or "emailAddress" not in body:
        raise BadRequest("Missing email address")

    requested_email_address = validate_email_address(str(body["emailAddress"]))

    return {"available": is_email_available(requested_email_address)}


def get_apiary_user(username: str) -> Union[Dict[str, Any], None]:
    """
    Fetch a user from Apiary by username. Returns None if the user does not exist.
    """
    apiary_user_response = apiary.get(  # type: ignore
        url=app.config["APIARY_URL"] + "/api/v1/users/" + username,
        headers={"Accept": "application/json"},
        params={"include": "roles"},
        timeout=(5, 5),
    )

    if apiary_user_response.status_code == 200:
        return apiary_user_response.json()["user"]  # type: ignore

    if apiary_user_response.status_code == 404:
        return None

    apiary_user_response.raise_for_status()
    return None


def get_manager_google_workspace_email(manager_uid: str) -> Union[str, None]:
    """
    Resolve a manager's Google Workspace email via Keycloak, if they have an account.
    """
    search_keycloak_user_response = keycloak.get(  # type: ignore
        url=keycloak_server + "/admin/realms/" + app.config["KEYCLOAK_REALM"] + "/users",
        params={
            "username": manager_uid,
            "exact": "true",
        },
        timeout=(5, 5),
    )
    search_keycloak_user_response.raise_for_status()

    keycloak_users = search_keycloak_user_response.json()

    if len(keycloak_users) != 1:
        return None

    manager_keycloak_user = keycloak_users[0]
    attributes = manager_keycloak_user.get("attributes")

    if attributes is None:
        return None

    google_workspace_accounts = attributes.get("googleWorkspaceAccount")

    if google_workspace_accounts is None or len(google_workspace_accounts) == 0:
        return None

    return google_workspace_accounts[0]  # type: ignore


def is_primary_team_project_manager(
    apiary_user: Dict[str, Any], primary_team: Dict[str, Any]
) -> bool:
    """
    Return True if the Apiary user is the project manager of their primary team.
    """
    apiary_team_response = apiary.get(  # type: ignore
        url=app.config["APIARY_URL"] + "/api/v1/teams/" + str(primary_team["id"]),
        headers={"Accept": "application/json"},
        params={"include": "projectManager"},
        timeout=(5, 5),
    )

    if apiary_team_response.status_code == 404:
        return False

    if apiary_team_response.status_code != 200:
        apiary_team_response.raise_for_status()
        return False

    project_manager = apiary_team_response.json()["team"].get("project_manager")

    if project_manager is None or project_manager.get("id") is None:
        return False

    return str(project_manager["id"]) == str(apiary_user["id"])


def has_apiary_admin_role(apiary_user: Dict[str, Any]) -> bool:
    """
    Return True if the Apiary user has the admin role.
    """
    roles = apiary_user.get("roles")

    if roles is None:
        return False

    for role in roles:
        if role.get("name") == "admin":
            return True

    return False


def build_google_workspace_user_body(
    first_name: str,
    last_name: str,
    email_address: str,
    apiary_user: Union[Dict[str, Any], None],
) -> Dict[str, Any]:
    """
    Build the Google Directory user insert body, including optional Apiary-backed fields.
    """
    body: Dict[str, Any] = {
        "name": {
            "givenName": first_name,
            "familyName": last_name,
        },
        "primaryEmail": email_address,
        "password": uuid4().hex,
    }

    if apiary_user is None:
        return body

    if has_apiary_admin_role(apiary_user):
        body["orgUnitPath"] = "/Super Administrators"

    if apiary_user.get("phone_verified") is True and apiary_user.get("phone"):
        body["phones"] = [
            {
                "value": apiary_user["phone"],
                "type": "mobile",
            }
        ]

    primary_team = apiary_user.get("primary_team")

    if (
        primary_team is not None
        and primary_team.get("id") is not None
        and primary_team.get("name") is not None
    ):
        organization: Dict[str, Any] = {
            "department": primary_team["name"],
            "primary": True,
        }

        if is_primary_team_project_manager(apiary_user, primary_team):
            organization["title"] = "Project Manager"

        body["organizations"] = [organization]

    manager = apiary_user.get("manager")

    if manager is not None and manager.get("uid") is not None:
        manager_email = get_manager_google_workspace_email(manager["uid"])

        if manager_email is not None:
            body["relations"] = [
                {
                    "value": manager_email,
                    "type": "manager",
                }
            ]

    return body


@app.post("/")
def submit() -> Any:
    """
    Create the Google Workspace account
    """
    if "user_state" not in session:
        raise Unauthorized("Not logged in")

    if session["user_state"] != "eligible":
        raise Unauthorized("Not eligible")

    set_user(
        {
            "id": session["sub"],
            "ip_address": request.remote_addr,
        }
    )

    if "first_name" not in request.form:
        raise BadRequest("Missing first name")

    if "last_name" not in request.form:
        raise BadRequest("Missing last name")

    if "email_address" not in request.form:
        raise BadRequest("Missing email address")

    first_name = validate_name("first", request.form["first_name"])
    last_name = validate_name("last", request.form["last_name"])
    email_address = validate_email_address(request.form["email_address"])

    if not is_email_available(email_address):
        raise Conflict("This email address isn't available")

    get_keycloak_user_response = keycloak.get(  # type: ignore
        url=keycloak_server
        + "/admin/realms/"
        + app.config["KEYCLOAK_REALM"]
        + "/users/"
        + session["sub"],
        timeout=(5, 5),
    )
    get_keycloak_user_response.raise_for_status()

    new_user = get_keycloak_user_response.json()
    keycloak_username = new_user.get("username")
    apiary_user = get_apiary_user(keycloak_username) if keycloak_username is not None else None

    new_workspace_user = google_workspace.insert(
        body=build_google_workspace_user_body(first_name, last_name, email_address, apiary_user),
        resolveConflictAccount=True,
    ).execute()

    cache.delete_memoized(is_email_available, email_address)

    if "id" in new_user:
        del new_user["id"]

    if "username" in new_user:
        del new_user["username"]

    if "attributes" not in new_user:
        new_user["attributes"] = {"googleWorkspaceAccount": [new_workspace_user["primaryEmail"]]}
    else:
        new_user["attributes"]["googleWorkspaceAccount"] = [new_workspace_user["primaryEmail"]]

    update_keycloak_user_response = keycloak.put(  # type: ignore
        url=keycloak_server
        + "/admin/realms/"
        + app.config["KEYCLOAK_REALM"]
        + "/users/"
        + session["sub"],
        json=new_user,
        timeout=(5, 5),
    )
    update_keycloak_user_response.raise_for_status()

    remove_eligible_role.delay(session["sub"])
    import_user_to_org_chart.delay(new_workspace_user["id"])
    notify_slack_account_created.delay(session["sub"])

    # gmail routing doesn't work immediately when a new account is created,
    # so we wait 60 seconds before sending the hubspot invitation and hope that's long enough
    invite_user_to_hubspot.apply_async((new_workspace_user["id"],), countdown=60)

    return redirect(
        "https://www.google.com/a/robojackets.org/ServiceLogin?continue=https://workspace.google.com/dashboard"  # noqa
    )


@app.post("/slack")
def handle_slack_event() -> Dict[str, str]:
    """
    Handle an interaction event from Slack

    https://docs.slack.dev/interactivity/handling-user-interaction#payloads
    """
    verifier = SignatureVerifier(app.config["SLACK_SIGNING_SECRET"])

    if not verifier.is_valid_request(request.get_data(), request.headers):  # type: ignore
        raise Unauthorized("signature verification failed")

    payload = loads(request.form.get("payload"))  # type: ignore

    if payload["actions"][0]["action_id"] == "view_in_apiary":
        return {"status": "ok"}

    if payload["actions"][0]["action_id"] == "view_in_workspace":
        return {"status": "ok"}

    if payload["actions"][0]["action_id"] == "view_in_keycloak":
        return {"status": "ok"}

    if payload["actions"][0]["action_id"] == "grant_eligibility_in_keycloak":
        add_eligible_role_response = keycloak.post(  # type: ignore
            url=keycloak_server
            + "/admin/realms/"
            + app.config["KEYCLOAK_REALM"]
            + "/users/"
            + str(UUID(payload["actions"][0]["value"]))
            + "/role-mappings/clients/"
            + app.config["KEYCLOAK_CLIENT_UUID"],
            timeout=(5, 5),
            json=[{"id": app.config["KEYCLOAK_CLIENT_ROLE_ELIGIBLE"], "name": "eligible"}],
        )
        add_eligible_role_response.raise_for_status()

        slack = WebhookClient(url=payload["response_url"])
        slack.send(
            text=payload["message"]["text"],
            blocks=[
                payload["message"]["blocks"][0],
                ActionsBlock(
                    elements=[
                        payload["message"]["blocks"][1]["elements"][0],
                        payload["message"]["blocks"][1]["elements"][1],
                    ]
                ),
                SectionBlock(
                    text=":white_check_mark: *<@"
                    + payload["user"]["id"]
                    + "> granted eligibility in Keycloak*"
                ),
            ],
            replace_original=True,
        )

        return {"status": "ok"}

    raise BadRequest("unrecognized action_id")


@app.get("/ping")
def ping() -> Dict[str, str]:
    """
    Returns an arbitrary successful response, for health checks
    """
    return {"status": "ok"}


@app.cli.command("clear-cache")
def clear_cache() -> None:
    """
    Clears the cache
    """
    cache.clear()
    print("Cache cleared.")