implement user define secret management api

aggregator/rpc_server.go

@@ -248,6 +248,77 @@ func (r *RpcServer) TriggerTask(ctx context.Context, payload *avsproto.UserTrigg
 	return r.engine.TriggerTask(user, payload)
 }
 
+func (r *RpcServer) CreateSecret(ctx context.Context, payload *avsproto.CreateOrUpdateSecretReq) (*wrapperspb.BoolValue, error) {
+	user, err := r.verifyAuth(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Unauthenticated, "%s: %s", auth.AuthenticationError, err.Error())
+	}
+
+	r.config.Logger.Info("process create secret",
+		"user", user.Address.String(),
+		"secret_name", payload.Name,
+	)
+
+	result, err := r.engine.CreateSecret(user, payload)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "")
+	}
+
+	return wrapperspb.Bool(result), nil
+}
+
+func (r *RpcServer) ListSecrets(ctx context.Context, payload *avsproto.ListSecretsReq) (*avsproto.ListSecretsResp, error) {
+	user, err := r.verifyAuth(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Unauthenticated, "%s: %s", auth.AuthenticationError, err.Error())
+	}
+
+	r.config.Logger.Info("process list secret",
+		"user", user.Address.String(),
+	)
+
+	return r.engine.ListSecrets(user, payload)
+}
+
+func (r *RpcServer) UpdateSecret(ctx context.Context, payload *avsproto.CreateOrUpdateSecretReq) (*wrapperspb.BoolValue, error) {
+	user, err := r.verifyAuth(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Unauthenticated, "%s: %s", auth.AuthenticationError, err.Error())
+	}
+
+	r.config.Logger.Info("process update secret",
+		"user", user.Address.String(),
+		"secret_name", payload.Name,
+	)
+
+	result, err := r.engine.UpdateSecret(user, payload)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "")
+	}
+
+	return wrapperspb.Bool(result), nil
+}
+
+func (r *RpcServer) DeleteSecret(ctx context.Context, payload *avsproto.DeleteSecretReq) (*wrapperspb.BoolValue, error) {
+	user, err := r.verifyAuth(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Unauthenticated, "%s: %s", auth.AuthenticationError, err.Error())
+	}
+
+	r.config.Logger.Info("process delete secret",
+		"user", user.Address.String(),
+		"secret_name", payload.Name,
+	)
+
+	result, err := r.engine.DeleteSecret(user, payload)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "")
+	}
+
+	return wrapperspb.Bool(result), nil
+
+}
+
 // Operator action
 func (r *RpcServer) SyncMessages(payload *avsproto.SyncMessagesReq, srv avsproto.Node_SyncMessagesServer) error {
 	err := r.engine.StreamCheckToOperator(payload, srv)

core/taskengine/doc.go

@@ -24,5 +24,10 @@ Storage can also be inspect with telnet:
 	telnet /tmp/ap.sock
 
 Then issue `get <ket>` or `list <prefix>` or `list *` to inspect current keys in the storage.
+
+**Secret Storage**
+- currently org_id will always be _ because we haven't implemented it yet
+- workflow_id will also be _ so we can search be prefix
+secret:<org_id>:<eoa>:<workflow_id>:<name> -> value
 */
 package taskengine

core/taskengine/engine.go

@@ -31,8 +31,9 @@ import (
 )
 
 const (
-	JobTypeExecuteTask = "execute_task"
-	DefaultItemPerPage = 50
+	JobTypeExecuteTask  = "execute_task"
+	DefaultItemPerPage  = 50
+	MaxSecretNameLength = 255
 )
 
 var (
@@ -849,6 +850,100 @@ func (n *Engine) CancelTaskByUser(user *model.User, taskID string) (bool, error)
 	return true, nil
 }
 
+func (n *Engine) CreateSecret(user *model.User, payload *avsproto.CreateOrUpdateSecretReq) (bool, error) {
+	secret := &model.Secret{
+		User:       user,
+		Name:       payload.Name,
+		Value:      payload.Secret,
+		OrgID:      payload.OrgId,
+		WorkflowID: payload.WorkflowId,
+	}
+
+	updates := map[string][]byte{}
+	if strings.HasPrefix(strings.ToLower(payload.Name), "ap_") {
+		return false, grpcstatus.Errorf(codes.InvalidArgument, "secret name cannot start with ap_")
+	}
+
+	if len(payload.Name) == 0 || len(payload.Name) > MaxSecretNameLength {
+		return false, grpcstatus.Errorf(codes.InvalidArgument, "secret name lengh is invalid: should be 1-255 character")
+	}
+
+	key, _ := SecretStorageKey(secret)
+	updates[key] = []byte(payload.Secret)
+	err := n.db.BatchWrite(updates)
+	if err == nil {
+		return true, nil
+	}
+
+	return false, grpcstatus.Errorf(codes.Internal, "Cannot save data")
+}
+
+func (n *Engine) UpdateSecret(user *model.User, payload *avsproto.CreateOrUpdateSecretReq) (bool, error) {
+	updates := map[string][]byte{}
+	secret := &model.Secret{
+		User:       user,
+		Name:       payload.Name,
+		Value:      payload.Secret,
+		OrgID:      payload.OrgId,
+		WorkflowID: payload.WorkflowId,
+	}
+	key, _ := SecretStorageKey(secret)
+	if ok, err := n.db.Exist([]byte(key)); !ok || err != nil {
+		return false, grpcstatus.Errorf(codes.NotFound, "Secret not found")
+	}
+
+	updates[key] = []byte(payload.Secret)
+
+	err := n.db.BatchWrite(updates)
+	if err == nil {
+		return true, nil
+	}
+
+	return true, nil
+}
+
+// ListSecrets
+func (n *Engine) ListSecrets(user *model.User, payload *avsproto.ListSecretsReq) (*avsproto.ListSecretsResp, error) {
+	prefixes := []string{
+		SecretStoragePrefix(user),
+	}
+
+	result := &avsproto.ListSecretsResp{
+		Items: []*avsproto.ListSecretsResp_ResponseSecret{},
+	}
+
+	secretKeys, err := n.db.ListKeysMulti(prefixes)
+	if err != nil {
+		return nil, err
+	}
+	for _, k := range secretKeys {
+		secretWithNameOnly := SecretNameFromKey(k)
+		item := &avsproto.ListSecretsResp_ResponseSecret{
+			Name:       secretWithNameOnly.Name,
+			OrgId:      secretWithNameOnly.OrgID,
+			WorkflowId: secretWithNameOnly.WorkflowID,
+		}
+
+		result.Items = append(result.Items, item)
+	}
+
+	return result, nil
+}
+
+func (n *Engine) DeleteSecret(user *model.User, payload *avsproto.DeleteSecretReq) (bool, error) {
+	// No need to check permission, the key is prefixed by user eoa already
+	secret := &model.Secret{
+		Name:       payload.Name,
+		User:       user,
+		OrgID:      payload.OrgId,
+		WorkflowID: payload.WorkflowId,
+	}
+	key, _ := SecretStorageKey(secret)
+	err := n.db.Delete([]byte(key))
+
+	return err == nil, err
+}
+
 // A global counter for the task engine
 func (n *Engine) NewSeqID() (string, error) {
 	num := uint64(0)