Repository: aipotheosis-labs/aci
File: backend/aci/server/app_connectors/agent_secrets_manager.py (Python)

Code before:

            return DomainCredential(
                domain=domain,
                **secret_value.model_dump(),
            )

    def create_credential_for_domain(self, domain: str, username: str, password: str) -> None:
        """
        Creates a new credential for a specific domain.

        Function name: AGENT_SECRETS_MANAGER__CREATE_CREDENTIAL_FOR_DOMAIN

        Args:
            domain (str): Domain for the credential.
            username (str): Username for the credential.
            password (str): Password for the credential.

        Raises:
            AgentSecretsManagerError: If a credential for the domain already exists.
            MaxAgentSecretsReached: If the project has reached its maximum allowed agent secrets
            SubscriptionPlanNotFound: If the organization's subscription plan cannot be found
            ProjectNotFound: If the project cannot be found
        """
        with create_db_session(config.DB_FULL_URL) as db_session:
            # Check quota before creating new secret
            quota_manager.enforce_agent_secrets_quota(db_session, self.linked_account.project_id)

            existing = crud.secret.get_secret(db_session, self.linked_account.id, domain)
            if existing:
                raise AgentSecretsManagerError(
                    message=f"Credential for domain '{domain}' already exists"
                )

            secret_value = SecretValue(username=username, password=password)
            encrypted_value = encryption.encrypt(secret_value.model_dump_json().encode())

            secret_create = SecretCreate(
                key=domain,
                value=encrypted_value,
            )
            crud.secret.create_secret(db_session, self.linked_account.id, secret_create)
            db_session.commit()

    def update_credential_for_domain(self, domain: str, username: str, password: str) -> None:
        """
        Updates an existing credential for a specific domain.

        Function name: AGENT_SECRETS_MANAGER__UPDATE_CREDENTIAL_FOR_DOMAIN

        Args:
            domain (str): Domain for the credential to update.

Review comment (security):
this check should be put after line `if existing:`
because creating duplicate secrets should be surfaced as the error, instead of giving quota error.

Code after:

            return DomainCredential(
                domain=domain,
                **secret_value.model_dump(),
            )

    def create_credential_for_domain(self, domain: str, username: str, password: str) -> None:
        """
        Creates a new credential for a specific domain.

        Function name: AGENT_SECRETS_MANAGER__CREATE_CREDENTIAL_FOR_DOMAIN

        Args:
            domain (str): Domain for the credential.
            username (str): Username for the credential.
            password (str): Password for the credential.

        Raises:
            AgentSecretsManagerError: If a credential for the domain already exists.
            MaxAgentSecretsReached: If the project has reached its maximum allowed agent secrets
            SubscriptionPlanNotFound: If the organization's subscription plan cannot be found
            ProjectNotFound: If the project cannot be found
        """
        with create_db_session(config.DB_FULL_URL) as db_session:
            existing = crud.secret.get_secret(db_session, self.linked_account.id, domain)
            if existing:
                raise AgentSecretsManagerError(
                    message=f"Credential for domain '{domain}' already exists"
                )

            # Check quota before creating new secret
            quota_manager.enforce_agent_secrets_quota(db_session, self.linked_account.project_id)

            secret_value = SecretValue(username=username, password=password)
            encrypted_value = encryption.encrypt(secret_value.model_dump_json().encode())

            secret_create = SecretCreate(
                key=domain,
                value=encrypted_value,
            )
            crud.secret.create_secret(db_session, self.linked_account.id, secret_create)
            db_session.commit()

    def update_credential_for_domain(self, domain: str, username: str, password: str) -> None:
        """
        Updates an existing credential for a specific domain.

        Function name: AGENT_SECRETS_MANAGER__UPDATE_CREDENTIAL_FOR_DOMAIN

        Args:
            domain (str): Domain for the credential to update.