cx-keys/assets/application_secret.py

from .user import user
from .user import user_builder
from .user_loader import user_loader
from .secret import secret
from .secret import secret_builder
from .secret_coder import secret_coder
from .secret_loader import secret_loader
from .application_part import application_part
from .apikey import bad_apikey_exception
from .validators import name_validator
from .validators import domain_validator
from .application_secret_response import secret_response
from .application_secret_response import secret_collection_response

class application_secret(application_part):
    """ 
    This class is responsible for managing secrets in the application. 
    """

    def get(self, apikey: str, name: str) -> dict:
        """
        This load secret by name. It return full secret, with crypted content, 
        or fail response when secret with given name not exists.

        Parameters:
            apikey (str): ApiKey of the secret owner
            name (str): Name of the secret

        Returns:
            (dict): Result to convert into JSON
        """

        with self.__secret_database(apikey) as loader:
            target = loader.load_by_name(name)

            if target is None:
                return self.__not_found_response()

            return secret_response(target).response

    def __validator(self, **kwargs) -> dict | None:
        """
        This validate given parameters. It return None when all given items 
        is valid, or dict with fail response dict.

        Kwargs:
            name (str): Name to valid
            new_name (str): Name to valid
            domain (str): Domain to valid
            coded (str): Crypted content to valid

        Returns:
            (dict): Fail to return as JSON
            (None): Validating parameters ara valid
        """
        
        result = None

        if "name" in kwargs:
            name = kwargs["name"]
            result = self._validation("name", name_validator(name))
        
        if "new_name" in kwargs:
            name = kwargs["new_name"]
            result = self._validation("name", name_validator(name))

        if "domain" in kwargs and result is None:
            domain = kwargs["domain"]
            result = self._validation("domain", domain_validator(domain))

        if not "coded" in kwargs:
            return result
            
        coded = kwargs["coded"]

        if not secret_coder.validate(coded):
            return self._fail_response(cause = "Invalid coded secret.")

        return result

    def create(self, apikey: str, name: str, domain: str, coded: str) -> dict:
        """
        This register new secret.

        Parameters:
            apikey (str): ApiKey of the owner
            name (str): Name of the secret
            domain (str): Domain to the secret
            coded (str): Crypted content of the secret

        Returns:
            (dict): Result of the action to convert into JSON
        """

        validation = self.__validator(
            name = name,
            domain = domain,
            coded = coded
        )

        if validation is not None:
            return validation

        with self.__secret_database(apikey) as loader:
            builder = secret_builder()
            builder.name = name
            builder.domain = domain
            builder.coded = coded
            builder.owner = loader.owner

            if loader.append(builder.result):   
                return self._success_response()

            return self._fail_response(cause = "Name already in use.")

    def drop(self, apikey: str, name: str) -> dict:
        """
        This remove secret from database.

        Parameters:
            apikey (str): Owner of the secret
            name (str): Name of the secret

        Returns:
            (dict): Result to return as JSON
        """

        with self.__secret_database(apikey) as loader:
            target = loader.load_by_name(name) 

            if target is None:
                return self.__not_found_response()

            if loader.drop(target):
                return self._success_response()

            return self._fail_response(cause = "Internal database error.")

    def update(self, apikey: str, name: str, **kwargs) -> dict:
        """
        This update secret data in the database.

        Parameters: 
            apikey (str): ApiKey of the secret owner
            name (str): Old name of the secret

        Kwargs:
            new_name (str): New name of the secret
            domain (str): Domain of the secret
            coded (str): Coded content of the secret 
        """

        validation = self.__validator(**kwargs)

        if validation is not None:
            return validation

        with self.__secret_database(apikey) as loader:
            target = loader.load_by_name(name)

            if target is None:
                return self.__not_found_response()

            builder = secret_builder(target)

            if "new_name" in kwargs:
                builder.name = kwargs["new_name"]

            if "domain" in kwargs:
                builder.domain = kwargs["domain"]

            if "coded" in kwargs:
                builder.coded = kwargs["coded"]

            if loader.update(builder.result):                
                return self._success_response()

            return self._fail_response(cause = "Name already in use.")

    def name_in_use(self, apikey: str, name: str) -> dict:
        """
        This check that name is used by any sectet.

        Parameters:
            apikey (str): ApiKey of the owner
            name (str): Name to check

        Returns:
            (dict): Result of the operation to return as JSON
        """

        validation = self._validation("name", name_validator(name))

        if validation is not None:
            return validation

        with self.__secret_database(apikey) as loader:
            result = loader.name_in_use(name)

            return self._success_response(in_use = result, name = name)

    def domain_search(self, apikey: str, domain: str) -> dict:
        """
        This search in database by domain name.

        Parameters:
            apikey (str): ApiKey of the secrets owner
            domain (str): Domain name to search

        Returns:
            (dict): Result to return as JSON
        """

        with self.__secret_database(apikey) as loader:
            results = loader.search_for_domain(domain)

            return secret_collection_response(results).response

    def name_search(self, apikey: str, name: str) -> dict:
        """
        This search in database by domain name.

        Parameters:
            apikey (str): ApiKey of the secrets owner
            name (str): Name to search

        Returns:
            (dict): Result to return as JSON
        """

        with self.__secret_database(apikey) as loader:
            results = loader.search_for_name(name) 

            return secret_collection_response(results).response

    def get_all(self, apikey: str) -> dict:
        """
        This return all secrets own by user.

        Parameters:
            apikey (str): ApiKey of the user

        Returns:
            (dict): Collection of the user secrets
        """
        
        with self.__secret_database(apikey) as loader:
            results = loader.get_all()

            return search_collection_response(results).response

    def __not_found_response(self) -> dict:
        """ 
        It create response with not found error.

        Returns:
            (dict): Result to return as JSON
        """

        return self._fail_response(cause = "Secret not found.")

    @property
    def __user_database(self) -> user_loader:
        """ It return user database. """
        
        return user_loader(self._connector)

    def __secret_database(self, apikey: str) -> secret_loader:
        """
        It return new secret database for user with given apikey. When
        user is not exists, then raise bad_apikey_exception.

        Parameters:
            apikey (str): ApiKey of the owner

        Returns:
            (secret_loader): Secret loader for the owner
        """

        with self.__user_database as loader:
            target = loader.get_by_apikey(apikey)

            if target is None:
                raise bad_apikey_exception()

            return secret_loader(self._connector, target)