openserverless/impl/auth/auth_service.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
import datetime
import json
import os
import logging
import secrets
import string
import time
import openserverless.common.response_builder as res_builder
import openserverless.couchdb.bcrypt_util as bu

from openserverless.common.oidc_validator import (
    OidcForbiddenError,
    OidcTokenValidator,
    OidcValidationError,
)
from openserverless.common.sso_namespace import SsoNamespaceMapper
from openserverless.couchdb.couchdb_util import CouchDB
from openserverless.common.kube_api_client import KubeApiClient

USER_META_DBN = "users_metadata"
SSO_PROVIDER_ANNOTATION = "openserverless.apache.org/auth-provider"
SSO_MODE_ANNOTATION = "openserverless.apache.org/auth-mode"
SSO_USERNAME_ANNOTATION = "openserverless.apache.org/sso-username"
SSO_SUB_ANNOTATION = "openserverless.apache.org/sso-sub"
SSO_ISSUER_ANNOTATION = "openserverless.apache.org/sso-issuer"
SSO_DISABLED_ANNOTATION = "openserverless.apache.org/sso-disabled"


class AuthService:

    def __init__(self, environ=os.environ, couch_db=None, kube_client=None):
        self._environ = environ
        self.couch_db = couch_db if couch_db is not None else CouchDB()
        self.kube_client = kube_client if kube_client is not None else KubeApiClient()

    def fetch_user_data(self, login: str):
        logging.info(f"searching for user {login} data")
        try:
            selector = {"selector": {"login": {"$eq": login}}}
            response = self.couch_db.find_doc(USER_META_DBN, json.dumps(selector))

            if response["docs"]:
                docs = list(response["docs"])
                if len(docs) > 0:
                    return docs[0]

            logging.warning(f"OpenServerless metadata for user {login} not found!")
            return None
        except Exception as e:
            logging.error(
                f"failed to query OpenServerless metadata for user {login}. Reason: {e}"
            )
            return None

    def env_to_dict(self, user_data, key="env"):
        """
        extract env from user_data and return it as a dict

        Keyword arguments:
        key -- the key to extract the env from
        """
        body = {}
        if key in user_data:
            envs = list(user_data[key])
        else:
            envs = []

        for env in envs:
            body[env['key']] = env['value']

        return body

    def map_data(self, user_data):
        """
        Map the internal openserverless user_data records to the auth response
        """
        resp = {}
        resp["login"] = user_data["login"]
        resp["email"] = user_data["email"]

        if "env" in user_data:
            resp["env"] = user_data["env"]

        if "quota" in user_data:
            resp["quota"] = user_data["quota"]

        return resp

    def login(self, login, password):
        user_data = self.fetch_user_data(login)

        if user_data:
            if bu.verify_password(password, user_data["password"]):
                # if(password == user_data['password']):
                return res_builder.build_response_with_data(self.map_data(user_data))
            else:
                logging.warning(f"password mismatch for user {login}")
                return res_builder.build_error_message(f"Invalid credentials", 401)
        else:
            logging.warning(f"no user {login} found")
            return res_builder.build_error_message(f"Invalid credentials", 401)

    def login_oidc(self, access_token, expected_namespace=None):
        try:
            validator = OidcTokenValidator(self._environ)
            claims = validator.validate(access_token)
        except OidcForbiddenError:
            return res_builder.build_error_message("Forbidden", 403)
        except OidcValidationError as exc:
            logging.warning(f"OIDC token validation failed: {exc}")
            return res_builder.build_error_message("Invalid OIDC token", 401)

        # A username is a routing hint, not a stable external identity.
        if any(not isinstance(claims.get(key), str) or not claims[key].strip()
               for key in ("iss", "sub")):
            return res_builder.build_error_message("Invalid OIDC identity", 401)

        username_claim = self._environ.get("OIDC_USERNAME_CLAIM", "preferred_username")
        external_username = claims[username_claim]
        try:
            login = SsoNamespaceMapper(self._environ).namespace_for(claims)
        except ValueError as exc:
            logging.warning(f"OIDC user {external_username} namespace mapping failed: {exc}")
            return res_builder.build_error_message("Invalid OIDC namespace mapping", 400)

        if expected_namespace and login != expected_namespace:
            logging.warning(
                f"OIDC user {external_username} resolved namespace {login} "
                f"does not match requested namespace {expected_namespace}"
            )
            return res_builder.build_error_message(
                "Authenticated namespace does not match requested workspace",
                403,
            )

        whisk_user = self.kube_client.get_whisk_user(login)
        if whisk_user and not self.is_sso_identity_bound(whisk_user, claims):
            logging.warning(f"OIDC identity is not bound to namespace {login}")
            return res_builder.build_error_message("Forbidden", 403)

        user_data = self.fetch_user_data(login)

        if user_data and not whisk_user:
            # Orphan metadata must not be claimed through a concurrent create.
            return res_builder.build_error_message("Forbidden", 403)

        if not user_data:
            user_data = self.provision_oidc_user_if_enabled(login, external_username, claims)

        if not user_data:
            logging.warning(f"OIDC user {external_username} has no provisioned namespace")
            return res_builder.build_error_message("Namespace not provisioned", 404)

        # Re-read after provisioning/waiting: an existing resource, including
        # one found after a create conflict, must belong to this exact identity.
        current_whisk_user = self.kube_client.get_whisk_user(login)
        if not self.is_sso_identity_bound(current_whisk_user, claims):
            return res_builder.build_error_message("Forbidden", 403)
        if whisk_user and (
            (whisk_user.get("metadata") or {}).get("uid")
            != (current_whisk_user.get("metadata") or {}).get("uid")
        ):
            return res_builder.build_error_message("Forbidden", 403)

        login_data = self.map_login_data(user_data)
        if "AUTH" not in login_data:
            logging.error(f"OIDC user {login} metadata is missing AUTH")
            return res_builder.build_error_message("Namespace metadata is missing AUTH", 500)

        login_data["LOGIN"] = login
        login_data["NAMESPACE"] = login
        return res_builder.build_response_with_data(login_data)

    def provision_oidc_user_if_enabled(self, login, external_username, claims):
        if not self._is_truthy(self._environ.get("SSO_AUTOPROVISION_ON_LOGIN")):
            return None

        email = claims.get("email")
        if not email:
            logging.warning(f"OIDC user {login} cannot be provisioned without email")
            return None

        if not self.ensure_whisk_user(login, external_username, email, claims):
            return None

        return self.wait_for_user_data(login)

    def ensure_whisk_user(self, login, external_username, email, claims):
        existing_whisk_user = self.kube_client.get_whisk_user(login)
        if existing_whisk_user:
            return self.is_sso_identity_bound(existing_whisk_user, claims)

        whisk_user = self.build_sso_whisk_user(login, external_username, email, claims)
        if self.kube_client.create_whisk_user(whisk_user):
            logging.info(f"WhiskUser for OIDC user {login} created")
            return True

        # A concurrent login may have created the CR between get and create.
        existing_whisk_user = self.kube_client.get_whisk_user(login)
        if existing_whisk_user:
            return self.is_sso_identity_bound(existing_whisk_user, claims)

        logging.error(f"failed to create WhiskUser for OIDC user {login}")
        return False

    def build_sso_whisk_user(self, login, external_username, email, claims):
        auth = self._random_auth()
        password = self._random_secret(24)
        services = self._environ.get("SSO_AUTOPROVISION_DEFAULT_SERVICES", "all")

        whisk_user = {
            "apiVersion": "openserverless.org/v1",
            "kind": "WhiskUser",
            "metadata": {
                "name": login,
                "namespace": "openserverless",
                "annotations": {
                    SSO_PROVIDER_ANNOTATION: self._environ.get("OIDC_PROVIDER", "keycloak"),
                    SSO_MODE_ANNOTATION: "sso",
                    SSO_USERNAME_ANNOTATION: external_username,
                    SSO_SUB_ANNOTATION: claims.get("sub", ""),
                    SSO_ISSUER_ANNOTATION: claims.get("iss", ""),
                },
            },
            "spec": {
                "email": email,
                "password": password,
                "namespace": login,
                "auth": auth,
            },
        }

        if services == "all":
            whisk_user["spec"]["redis"] = {
                "enabled": True,
                "prefix": login,
                "password": self._random_secret(12),
            }
            whisk_user["spec"]["mongodb"] = {
                "enabled": True,
                "database": login,
                "password": self._random_secret(12),
            }
            whisk_user["spec"]["postgres"] = {
                "enabled": True,
                "database": login,
                "password": self._random_secret(12),
            }
            whisk_user["spec"]["object-storage"] = {
                "password": self._random_secret(40),
                "quota": self._environ.get("SSO_AUTOPROVISION_STORAGE_QUOTA", "auto"),
                "data": {"enabled": True, "bucket": f"{login}-data"},
                "route": {"enabled": True, "bucket": f"{login}-web"},
            }
            whisk_user["spec"]["milvus"] = {
                "enabled": True,
                "database": login,
                "password": self._random_secret(12),
            }

        return whisk_user

    def wait_for_user_data(self, login):
        timeout_seconds = self._int_env("SSO_AUTOPROVISION_TIMEOUT_SECONDS", 120)
        poll_seconds = self._float_env("SSO_AUTOPROVISION_POLL_SECONDS", 2)
        deadline = time.monotonic() + timeout_seconds

        while True:
            user_data = self.fetch_user_data(login)
            if user_data:
                return user_data

            if time.monotonic() >= deadline:
                logging.warning(f"timeout waiting for OIDC namespace {login}")
                return None

            time.sleep(poll_seconds)

    def _random_auth(self):
        return f"{self._random_uuid()}:{self._random_secret(64)}"

    def _random_uuid(self):
        import uuid
        return str(uuid.uuid4())

    def _random_secret(self, length):
        alphabet = string.ascii_letters + string.digits
        return "".join(secrets.choice(alphabet) for _ in range(length))

    def _is_truthy(self, value):
        return str(value or "").lower() in ["1", "true", "yes", "on"]

    def _int_env(self, name, default):
        try:
            return int(self._environ.get(name, default))
        except (TypeError, ValueError):
            return default

    def _float_env(self, name, default):
        try:
            return float(self._environ.get(name, default))
        except (TypeError, ValueError):
            return default

    def is_sso_identity_bound(self, whisk_user, claims):
        if not whisk_user:
            return False

        annotations = ((whisk_user or {}).get("metadata") or {}).get("annotations") or {}
        return (
            not self._is_truthy(annotations.get(SSO_DISABLED_ANNOTATION))
            and isinstance(claims.get("iss"), str) and bool(claims["iss"].strip())
            and isinstance(claims.get("sub"), str) and bool(claims["sub"].strip())
            and annotations.get(SSO_ISSUER_ANNOTATION) == claims["iss"]
            and annotations.get(SSO_SUB_ANNOTATION) == claims["sub"]
        )

    def map_login_data(self, user_data):
        """
        Map user metadata to the flat string-only shape consumed by ops -login.
        This mirrors the whisk-system/nuv/login system action response.
        """
        body = self.env_to_dict(user_data, "userenv")

        for key, value in self.env_to_dict(user_data, "env").items():
            if key not in body:
                body[key] = value
            else:
                body[f"SYSTEM_{key}"] = value

        return body

    def update_password(self, login, old_password, new_password):
        user_data = self.fetch_user_data(login)

        if user_data:
            if bu.verify_password(old_password, user_data["password"]):
                whisk_user = self.kube_client.get_whisk_user(user_data["login"])

                whisk_user["spec"]["password"] = new_password
                # whisk_user['spec']['password_timestamp'] = datetime.now().isoformat()
                self.kube_client.update_whisk_user(whisk_user)

                return res_builder.build_response_with_data(
                    {"status": "ok", "message": "Password updated"}
                )
            else:
                return res_builder.build_error_message(
                    f"password mismatch for user {login}", 401
                )
        else:
            return res_builder.build_error_message(f"no user {login} found", 401)