openserverless/userdb_util.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 json
import logging

import openserverless.couchdb as cdb
import openserverless.couchdb_util as couchdb_util
import openserverless.util as util
from openserverless.openserverless_metadata import OpenServerlessMetadata
from openserverless.user_metadata import UserMetadata
import openserverless.bcrypt_util as bu
USER_META_DBN = "users_metadata"


def _add_metadata(db, metadata):
    """
    Add a new Openwhisk User metadata entry     
    """
    res = util.check(db.wait_db_ready(60), "wait_db_ready", True)
    return util.check(cdb.update_templated_doc(db, USER_META_DBN, "user_metadata.json", metadata),
                      f"add_metadata {metadata['login']}", res)


def save_user_metadata(user_metadata: UserMetadata):
    """
    Add a generic user metadata into the internal CouchDB 
    """
    metadata = user_metadata.get_metadata()
    logging.info(f"Storing OpenServerless metadata for {metadata['login']}")

    try:
        db = couchdb_util.CouchDB()
        return _add_metadata(db, metadata)
    except Exception as e:
        logging.error(f"failed to store OpenServerless metadata for {metadata['login']}. Cause: {e}")
        return None


def save_openserverless_metadata(openserverless_metadata: OpenServerlessMetadata):
    """
    Add openserverless user metadata into the internal CouchDB 
    """
    metadata = openserverless_metadata.get_metadata()
    logging.info(f"Storing OpenServerless metadata for {metadata['login']}")

    try:
        db = couchdb_util.CouchDB()
        return _add_metadata(db, metadata)
    except Exception as e:
        logging.error(f"failed to store OpenServerless metadata for {metadata['login']}. Cause: {e}")
        return None


def delete_user_metadata(login):
    logging.info(f"removing OpenServerless metadata for user {login}")

    try:
        db = couchdb_util.CouchDB()
        selector = {"selector": {"login": {"$eq": login}}}
        response = db.find_doc(USER_META_DBN, json.dumps(selector))

        if (response['docs']):
            docs = list(response['docs'])
            if (len(docs) > 0):
                doc = docs[0]
                logging.info(f"removing user metadata documents {doc['_id']}")
                return db.delete_doc(USER_META_DBN, doc['_id'])

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


def update_user_metadata_password(login, password):
    logging.info(f"updating OpenServerless password for user {login}")

    try:
        db = couchdb_util.CouchDB()
        selector = {"selector": {"login": {"$eq": login}}}
        response = db.find_doc(USER_META_DBN, json.dumps(selector))

        if (response['docs']):
            docs = list(response['docs'])
            if (len(docs) > 0):
                doc = docs[0]
                logging.info(f"updating user credentials {doc['_id']}")
                doc['password'] = bu.hash_password(password)
                return db.update_doc(USER_META_DBN, doc)

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


def update_user_metadata_quota(login, quota: list):
    logging.info(f"updating OpenServerless quota for user {login}")

    try:
        db = couchdb_util.CouchDB()
        selector = {"selector": {"login": {"$eq": login}}}
        response = db.find_doc(USER_META_DBN, json.dumps(selector))

        if (response['docs']):
            docs = list(response['docs'])
            if (len(docs) > 0):
                doc = docs[0]
                logging.info(f"updating user quota {doc['_id']}")
                doc['quota'] = quota
                return db.update_doc(USER_META_DBN, doc)

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