diff --git a/docker/.env b/docker/.env index 2e18fb8921..f70a65e3fa 100644 --- a/docker/.env +++ b/docker/.env @@ -493,6 +493,11 @@ DOTNET_SYSTEM_GLOBALIZATION_INVARIANT=1 # RAGFLOW_CRYPTO_ALGORITHM=aes-256-cbc # one of aes-256-cbc, aes-128-cbc, sm4-cbc # RAGFLOW_CRYPTO_KEY=ragflow-crypto-key +# Encrypts the credentials of data source connectors in the database. +# Generate it with `openssl rand -base64 32`. If the key is lost, stored credentials +# cannot be read: enter them again, or delete and re-create the connector. +# RAGFLOW_CONNECTOR_KEY= + # Used for ThreadPoolExecutor THREAD_POOL_MAX_WORKERS=128 diff --git a/docker/README.md b/docker/README.md index a7175dd3d1..2b5ce47162 100644 --- a/docker/README.md +++ b/docker/README.md @@ -153,6 +153,8 @@ The Go services use NATS JetStream for ingestion, synchronization, memory, and k Set to `true` only for development checkouts that intentionally use a development migration marker. Keep it `false` in production. - `RAGFLOW_IMAGE` The Go Docker image used by `docker-compose.yml`. Use an official Go image tag documented in its release notes, or use the locally built `ragflow:go-local` tag. The RAGFlow Docker image does not include embedding models. +- `RAGFLOW_CONNECTOR_KEY` + The key that encrypts the credentials of data source connectors in the database. Use 32 random bytes in standard base64, for example the output of `openssl rand -base64 32`. Not set by default, and then credentials are stored as plaintext. When the key is set, the database migration encrypts the credentials that are already stored. If the key is lost, stored credentials cannot be read: enter them again through the connector update API, or delete and re-create the connector. ### Local embedding service diff --git a/docs/administrator/setup/configuration.md b/docs/administrator/setup/configuration.md index b9f8ab5de5..e97c42837a 100644 --- a/docs/administrator/setup/configuration.md +++ b/docs/administrator/setup/configuration.md @@ -122,6 +122,8 @@ NATS JetStream provides the message queue used by the Go services. The public Nginx ports. Defaults to `80` and `443`. - `RAGFLOW_IMAGE` The Go RAGFlow image used by `docker-compose.yml`. Select the official Go image for the required release, or use the locally built `ragflow:go-local` image. +- `RAGFLOW_CONNECTOR_KEY` + The key that encrypts the credentials of data source connectors in the database. Use 32 random bytes in standard base64, for example the output of `openssl rand -base64 32`. Not set by default, and then credentials are stored as plaintext. When the key is set, the database migration encrypts the credentials that are already stored. If the key is lost, stored credentials cannot be read: enter them again through the connector update API, or delete and re-create the connector. :::tip NOTE If you cannot download the RAGFlow Docker image, try the following mirrors. diff --git a/docs/administrator/upgrade_and_migration/database_change_log.md b/docs/administrator/upgrade_and_migration/database_change_log.md index 9f201e684f..2f2bcecec2 100644 --- a/docs/administrator/upgrade_and_migration/database_change_log.md +++ b/docs/administrator/upgrade_and_migration/database_change_log.md @@ -357,3 +357,4 @@ and confirm free disk space. The existing `pipeline_title` index is preserved. | Table | Change | |---------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `knowledgebase`, `dialog` | `parser_config` entries stored under the legacy key `TokenChunker:SixApplesFall` are rewritten to `GeneralChunker:SixApplesFall`. Values already stored under the new key win; `TokenChunker`-only fields (`delimiter_mode`, `outputs`) are dropped, and a legacy `delimiter` becomes `delimiters`. `delimiter` is discarded when `delimiters` already exists. | +| `connector` | When `RAGFLOW_CONNECTOR_KEY` is set, a plaintext `config["credentials"]` value is encrypted to an `enc:v1:` string (AES-256-GCM). Rows that are already encrypted and rows whose `config` is not valid JSON are left unchanged. Without the key, the step changes nothing. | diff --git a/internal/common/connector_credentials.go b/internal/common/connector_credentials.go new file mode 100644 index 0000000000..0b018d5527 --- /dev/null +++ b/internal/common/connector_credentials.go @@ -0,0 +1,154 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed 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. +// + +package common + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "strings" +) + +// Stored rows use this format, so a change needs a new prefix version. +const ( + connectorCredentialsField = "credentials" + connectorCredentialsPrefix = "enc:v1:" + connectorCredentialsKeyBytes = 32 +) + +// Errors never carry the key, the ciphertext or the plaintext, because they reach API responses and task logs. +var ( + errConnectorKeyNotSet = errors.New("connector credentials are encrypted but " + EnvRAGFlowConnectorKey + " is not set") + errConnectorDecrypt = errors.New("cannot decrypt connector credentials: wrong " + EnvRAGFlowConnectorKey + " or corrupted value") +) + +// ConnectorKey returns the AES-256 key in RAGFLOW_CONNECTOR_KEY, or nil when the variable is unset or empty. +func ConnectorKey() ([]byte, error) { + encoded := GetEnv(EnvRAGFlowConnectorKey) + if encoded == "" { + return nil, nil + } + // Go's decoder skips \r and \n; reject them so only the documented one-line base64 form is accepted. + key, err := base64.StdEncoding.DecodeString(encoded) + if err != nil || strings.ContainsAny(encoded, "\r\n") { + return nil, fmt.Errorf("%s is not valid base64", EnvRAGFlowConnectorKey) + } + if len(key) != connectorCredentialsKeyBytes { + return nil, fmt.Errorf("%s must decode to %d bytes, got %d", EnvRAGFlowConnectorKey, connectorCredentialsKeyBytes, len(key)) + } + return key, nil +} + +// EncryptConnectorCredentials returns a copy of config with config["credentials"] encrypted. +// It returns config itself when the key is unset, credentials are absent, or they are already encrypted. +func EncryptConnectorCredentials(config map[string]interface{}) (map[string]interface{}, error) { + credentials, ok := config[connectorCredentialsField] + if !ok { + return config, nil + } + if value, isString := credentials.(string); isString && strings.HasPrefix(value, connectorCredentialsPrefix) { + return config, nil + } + key, err := ConnectorKey() + if err != nil { + return nil, err + } + if key == nil { + return config, nil + } + + plaintext, err := json.Marshal(credentials) + if err != nil { + return nil, fmt.Errorf("cannot encode connector credentials: %w", err) + } + gcm, err := connectorCredentialsCipher(key) + if err != nil { + return nil, err + } + nonce := make([]byte, gcm.NonceSize()) + if _, err := rand.Read(nonce); err != nil { + return nil, fmt.Errorf("cannot generate connector credentials nonce: %w", err) + } + sealed := gcm.Seal(nonce, nonce, plaintext, nil) + return withConnectorCredentials(config, connectorCredentialsPrefix+base64.StdEncoding.EncodeToString(sealed)), nil +} + +// DecryptConnectorCredentials returns a copy of config with encrypted credentials decoded. +// It returns config itself when the credentials are not encrypted. +func DecryptConnectorCredentials(config map[string]interface{}) (map[string]interface{}, error) { + value, ok := config[connectorCredentialsField].(string) + if !ok || !strings.HasPrefix(value, connectorCredentialsPrefix) { + return config, nil + } + key, err := ConnectorKey() + if err != nil { + return nil, err + } + if key == nil { + return nil, errConnectorKeyNotSet + } + + sealed, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(value, connectorCredentialsPrefix)) + if err != nil { + return nil, errConnectorDecrypt + } + gcm, err := connectorCredentialsCipher(key) + if err != nil { + return nil, err + } + if len(sealed) < gcm.NonceSize()+gcm.Overhead() { + return nil, errConnectorDecrypt + } + plaintext, err := gcm.Open(nil, sealed[:gcm.NonceSize()], sealed[gcm.NonceSize():], nil) + if err != nil { + return nil, errConnectorDecrypt + } + var credentials interface{} + if err := json.Unmarshal(plaintext, &credentials); err != nil { + return nil, errConnectorDecrypt + } + return withConnectorCredentials(config, credentials), nil +} + +// HasEncryptedConnectorCredentials reports whether config carries credentials in the stored "enc:" form. +// Requests must send plaintext: the server would store such a value as is and could not read it back. +func HasEncryptedConnectorCredentials(config map[string]interface{}) bool { + value, ok := config[connectorCredentialsField].(string) + return ok && strings.HasPrefix(value, "enc:") +} + +func connectorCredentialsCipher(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, err + } + return cipher.NewGCM(block) +} + +// withConnectorCredentials copies config so callers never see their map change. +func withConnectorCredentials(config map[string]interface{}, credentials interface{}) map[string]interface{} { + updated := make(map[string]interface{}, len(config)) + for key, value := range config { + updated[key] = value + } + updated[connectorCredentialsField] = credentials + return updated +} diff --git a/internal/common/connector_credentials_test.go b/internal/common/connector_credentials_test.go new file mode 100644 index 0000000000..78f41176ef --- /dev/null +++ b/internal/common/connector_credentials_test.go @@ -0,0 +1,207 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed 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. +// + +package common + +import ( + "bytes" + "encoding/base64" + "reflect" + "strings" + "testing" +) + +// Fixed public test vector. A format change that breaks it also breaks the rows that are already stored. +const ( + testConnectorKey = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" + testConnectorCipherVector = "enc:v1:ZGVmZ2hpamtsbW5vMzm/FhC2IvFVBzHK4EVIiS2pKzu5X9Feh/PZO57Rh3K0yyGkLTDmruUEeZB6LtRoLedehJBJUg==" +) + +func TestConnectorKey(t *testing.T) { + t.Setenv(EnvRAGFlowConnectorKey, "") + key, err := ConnectorKey() + if err != nil || key != nil { + t.Fatalf("unset key: got (%v, %v), want (nil, nil)", key, err) + } + + t.Setenv(EnvRAGFlowConnectorKey, testConnectorKey) + key, err = ConnectorKey() + if err != nil { + t.Fatalf("valid key: %v", err) + } + want := make([]byte, 32) + for i := range want { + want[i] = byte(i) + } + if !bytes.Equal(key, want) { + t.Fatalf("valid key: got %v, want %v", key, want) + } + + // Line breaks must fail, although Go's decoder skips them. + for _, bad := range []string{"not-base64-!!", base64.StdEncoding.EncodeToString(make([]byte, 16)), testConnectorKey + "\n", testConnectorKey + "\r\n", testConnectorKey + "\r", testConnectorKey[:20] + "\n" + testConnectorKey[20:]} { + t.Setenv(EnvRAGFlowConnectorKey, bad) + key, err = ConnectorKey() + if err == nil || key != nil { + t.Fatalf("bad key %q: got (%v, %v), want an error", bad, key, err) + } + if !strings.Contains(err.Error(), EnvRAGFlowConnectorKey) || strings.Contains(err.Error(), strings.TrimSpace(bad)) { + t.Fatalf("bad key %q: error %q must name the variable and not the value", bad, err) + } + } +} + +func TestConnectorCredentialsRoundTrip(t *testing.T) { + t.Setenv(EnvRAGFlowConnectorKey, testConnectorKey) + input := map[string]interface{}{ + "credentials": map[string]interface{}{"api_token": "tok-123"}, + "batch_size": 5, + } + + encrypted, err := EncryptConnectorCredentials(input) + if err != nil { + t.Fatalf("encrypt: %v", err) + } + stored, ok := encrypted["credentials"].(string) + if !ok || !strings.HasPrefix(stored, "enc:v1:") || strings.Contains(stored, "tok-123") { + t.Fatalf("encrypted credentials = %#v, want an enc:v1: string", encrypted["credentials"]) + } + if encrypted["batch_size"] != 5 { + t.Fatalf("batch_size = %#v, want 5", encrypted["batch_size"]) + } + if _, ok := input["credentials"].(map[string]interface{}); !ok { + t.Fatalf("encrypt mutated its input: %#v", input) + } + + again, err := EncryptConnectorCredentials(encrypted) + if err != nil || again["credentials"] != stored { + t.Fatalf("encrypting an encrypted config: got (%#v, %v), want it unchanged", again, err) + } + + decrypted, err := DecryptConnectorCredentials(encrypted) + if err != nil { + t.Fatalf("decrypt: %v", err) + } + want := map[string]interface{}{ + "credentials": map[string]interface{}{"api_token": "tok-123"}, + "batch_size": 5, + } + if !reflect.DeepEqual(decrypted, want) { + t.Fatalf("decrypted = %#v, want %#v", decrypted, want) + } + if encrypted["credentials"] != stored { + t.Fatalf("decrypt mutated its input: %#v", encrypted) + } +} + +func TestDecryptConnectorCredentialsFixedVector(t *testing.T) { + t.Setenv(EnvRAGFlowConnectorKey, testConnectorKey) + got, err := DecryptConnectorCredentials(map[string]interface{}{"credentials": testConnectorCipherVector, "wiki": "x"}) + if err != nil { + t.Fatalf("decrypt: %v", err) + } + want := map[string]interface{}{ + "credentials": map[string]interface{}{"api_token": "tok-123", "user": "ada"}, + "wiki": "x", + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("decrypted = %#v, want %#v", got, want) + } +} + +func TestDecryptConnectorCredentialsErrors(t *testing.T) { + sealed, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(testConnectorCipherVector, "enc:v1:")) + if err != nil { + t.Fatal(err) + } + sealed[len(sealed)-1] ^= 0x01 + tampered := "enc:v1:" + base64.StdEncoding.EncodeToString(sealed) + wrongKey := base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{7}, 32)) + + cases := []struct { + name string + key string + value string + want string + }{ + {"no key", "", testConnectorCipherVector, "connector credentials are encrypted but RAGFLOW_CONNECTOR_KEY is not set"}, + {"wrong key", wrongKey, testConnectorCipherVector, "cannot decrypt connector credentials"}, + {"tampered", testConnectorKey, tampered, "cannot decrypt connector credentials"}, + {"bad base64", testConnectorKey, "enc:v1:%%%", "cannot decrypt connector credentials"}, + {"too short", testConnectorKey, "enc:v1:" + base64.StdEncoding.EncodeToString([]byte("short")), "cannot decrypt connector credentials"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Setenv(EnvRAGFlowConnectorKey, tc.key) + got, err := DecryptConnectorCredentials(map[string]interface{}{"credentials": tc.value}) + if err == nil { + t.Fatalf("got %#v, want an error", got) + } + msg := err.Error() + if !strings.Contains(msg, tc.want) { + t.Fatalf("error %q does not contain %q", msg, tc.want) + } + for _, secret := range []string{testConnectorKey, wrongKey, "tok-123", strings.TrimPrefix(tc.value, "enc:v1:")} { + if strings.Contains(msg, secret) { + t.Fatalf("error %q leaks %q", msg, secret) + } + } + }) + } +} + +func TestConnectorCredentialsNothingToDo(t *testing.T) { + plain := map[string]interface{}{"credentials": map[string]interface{}{"api_token": "tok-123"}} + + t.Setenv(EnvRAGFlowConnectorKey, "") + got, err := EncryptConnectorCredentials(plain) + if err != nil || !reflect.DeepEqual(got, plain) { + t.Fatalf("encrypt without key: got (%#v, %v), want the input", got, err) + } + + t.Setenv(EnvRAGFlowConnectorKey, testConnectorKey) + noCredentials := map[string]interface{}{"wiki": "x"} + got, err = EncryptConnectorCredentials(noCredentials) + if err != nil || !reflect.DeepEqual(got, noCredentials) { + t.Fatalf("encrypt without credentials: got (%#v, %v), want the input", got, err) + } + got, err = DecryptConnectorCredentials(plain) + if err != nil || !reflect.DeepEqual(got, plain) { + t.Fatalf("decrypt plaintext: got (%#v, %v), want the input", got, err) + } + got, err = DecryptConnectorCredentials(nil) + if err != nil || got != nil { + t.Fatalf("decrypt nil: got (%#v, %v), want (nil, nil)", got, err) + } +} + +func TestHasEncryptedConnectorCredentials(t *testing.T) { + cases := []struct { + config map[string]interface{} + want bool + }{ + {map[string]interface{}{"credentials": testConnectorCipherVector}, true}, + {map[string]interface{}{"credentials": "enc:v2:abc"}, true}, + {map[string]interface{}{"credentials": "plain-token"}, false}, + {map[string]interface{}{"credentials": map[string]interface{}{"token": "enc:v1:x"}}, false}, + {map[string]interface{}{"wiki": "enc:v1:x"}, false}, + {nil, false}, + } + for _, tc := range cases { + if got := HasEncryptedConnectorCredentials(tc.config); got != tc.want { + t.Fatalf("HasEncryptedConnectorCredentials(%#v) = %v, want %v", tc.config, got, tc.want) + } + } +} diff --git a/internal/common/environments.go b/internal/common/environments.go index a218907451..82a6d1b7ff 100644 --- a/internal/common/environments.go +++ b/internal/common/environments.go @@ -264,6 +264,7 @@ const ( EnvFirecrawlTimeout = "FIRECRAWL_TIMEOUT" EnvFirecrawlRelayTimeout = "FIRECRAWL_RELAY_TIMEOUT" EnvRAGFlowSecretKey = "RAGFLOW_SECRET_KEY" + EnvRAGFlowConnectorKey = "RAGFLOW_CONNECTOR_KEY" EnvEnableRegister = "ENABLE_REGISTER" EnvDisablePasswordLogin = "DISABLE_PASSWORD_LOGIN" EnvMinioHost = "MINIO_HOST" diff --git a/internal/dao/connector_credentials_migration.go b/internal/dao/connector_credentials_migration.go new file mode 100644 index 0000000000..fb91b132f5 --- /dev/null +++ b/internal/dao/connector_credentials_migration.go @@ -0,0 +1,102 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed 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. + +package dao + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "ragflow/internal/common" + + "go.uber.org/zap" + "gorm.io/gorm" +) + +const connectorCredentialsMigrationBatchSize = 256 + +// The raw column text is read so the update can use it as an optimistic guard. +type connectorCredentialsMigrationRow struct { + ID string `gorm:"column:id"` + Config string `gorm:"column:config"` +} + +// migrateConnectorCredentials encrypts the plaintext credentials of existing +// connectors when RAGFLOW_CONNECTOR_KEY is set. A second run changes nothing. +func migrateConnectorCredentials(ctx context.Context, db *gorm.DB) error { + key, err := common.ConnectorKey() + if err != nil { + return err + } + if key == nil || !db.WithContext(ctx).Migrator().HasTable("connector") { + return nil + } + + var migrated int + var rows []connectorCredentialsMigrationRow + result := db.WithContext(ctx).Table("connector").Select("id", "config"). + FindInBatches(&rows, connectorCredentialsMigrationBatchSize, func(_ *gorm.DB, _ int) error { + for _, row := range rows { + // The whole row is rewritten, so numbers must survive exactly; float64 rounds above 2^53. + // Decode stops after the first value, so json.Valid keeps rows with trailing data skipped. + if !json.Valid([]byte(row.Config)) { + continue + } + var config map[string]interface{} + dec := json.NewDecoder(strings.NewReader(row.Config)) + dec.UseNumber() + if err := dec.Decode(&config); err != nil { + continue + } + if _, ok := config["credentials"]; !ok || common.HasEncryptedConnectorCredentials(config) { + continue + } + encrypted, err := common.EncryptConnectorCredentials(config) + if err != nil { + return fmt.Errorf("encrypt connector %q credentials: %w", row.ID, err) + } + value, err := json.Marshal(encrypted) + if err != nil { + return fmt.Errorf("encode connector %q config: %w", row.ID, err) + } + updated, err := updateConnectorConfigIfUnchanged(ctx, db, row.ID, row.Config, string(value)) + if err != nil { + return fmt.Errorf("update connector %q config: %w", row.ID, err) + } + if updated { + migrated++ + } + } + return nil + }) + if result.Error != nil { + return result.Error + } + if migrated > 0 { + common.Info("Encrypted connector credentials", zap.Int("connectors", migrated)) + } + return nil +} + +// updateConnectorConfigIfUnchanged writes newConfig only while the row still +// holds oldConfig, so it never overwrites a change made after the read. +func updateConnectorConfigIfUnchanged(ctx context.Context, db *gorm.DB, id, oldConfig, newConfig string) (bool, error) { + result := db.WithContext(ctx).Table("connector"). + Where("id = ? AND config = ?", id, oldConfig). + Update("config", newConfig) + return result.RowsAffected == 1, result.Error +} diff --git a/internal/dao/connector_credentials_migration_test.go b/internal/dao/connector_credentials_migration_test.go new file mode 100644 index 0000000000..06181a12f8 --- /dev/null +++ b/internal/dao/connector_credentials_migration_test.go @@ -0,0 +1,248 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed 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. + +package dao + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "encoding/base64" + "encoding/json" + "fmt" + "reflect" + "strings" + "testing" + + "gorm.io/gorm" + + "ragflow/internal/common" + "ragflow/internal/entity" +) + +const testConnectorKey = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" + +func setupConnectorCredentialsTestDB(t *testing.T) *gorm.DB { + t.Helper() + db := setupSyncTaskTestDB(t) + if err := db.AutoMigrate(&entity.Connector{}); err != nil { + t.Fatalf("failed to migrate Connector: %v", err) + } + return db +} + +func insertRawConnector(t *testing.T, db *gorm.DB, id, config string) { + t.Helper() + if err := db.Exec( + "INSERT INTO connector (id, tenant_id, name, source, input_type, config, status) VALUES (?, ?, ?, ?, ?, ?, ?)", + id, "tenant-1", id, "github", "poll", config, "schedule", + ).Error; err != nil { + t.Fatalf("insert connector %q: %v", id, err) + } +} + +func rawConnectorConfig(t *testing.T, db *gorm.DB, id string) string { + t.Helper() + var config string + if err := db.Raw("SELECT config FROM connector WHERE id = ?", id).Scan(&config).Error; err != nil { + t.Fatalf("read connector %q: %v", id, err) + } + return config +} + +func TestMigrateConnectorCredentialsEncryptsPlaintextRows(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsTestDB(t) + ctx := context.Background() + // The spacing differs from json.Marshal output, so any needless rewrite shows. + untouched := map[string]string{ + "no-credentials": `{"wiki": "x"}`, + "invalid-json": `not json`, + "trailing-data": `{"credentials": {"api_token": "tok-123"}} x`, + "encrypted": `{"credentials": "enc:v1:abc"}`, + } + insertRawConnector(t, db, "plain", `{"credentials":{"api_token":"tok-123"},"wiki":"x"}`) + for id, config := range untouched { + insertRawConnector(t, db, id, config) + } + + if err := migrateConnectorCredentials(ctx, db); err != nil { + t.Fatalf("migrate: %v", err) + } + + migrated := rawConnectorConfig(t, db, "plain") + var stored map[string]interface{} + if err := json.Unmarshal([]byte(migrated), &stored); err != nil { + t.Fatalf("migrated config %q is not JSON: %v", migrated, err) + } + if credentials, _ := stored["credentials"].(string); !strings.HasPrefix(credentials, "enc:v1:") { + t.Fatalf("migrated config = %s, want enc:v1: credentials", migrated) + } + decrypted, err := common.DecryptConnectorCredentials(stored) + if err != nil { + t.Fatalf("decrypt migrated config: %v", err) + } + want := map[string]interface{}{"credentials": map[string]interface{}{"api_token": "tok-123"}, "wiki": "x"} + if !reflect.DeepEqual(decrypted, want) { + t.Fatalf("decrypted = %#v, want %#v", decrypted, want) + } + for id, config := range untouched { + if got := rawConnectorConfig(t, db, id); got != config { + t.Fatalf("connector %q config = %q, want it untouched (%q)", id, got, config) + } + } + + if err := migrateConnectorCredentials(ctx, db); err != nil { + t.Fatalf("second migrate: %v", err) + } + if got := rawConnectorConfig(t, db, "plain"); got != migrated { + t.Fatalf("second run rewrote the row: %q, want %q", got, migrated) + } +} + +func TestMigrateConnectorCredentialsKeepsLargeIntegers(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsTestDB(t) + // Both numbers are above 2^53, where a float64 round trip changes them. + insertRawConnector(t, db, "c1", `{"credentials":{"app_id":9007199254740993},"channel_id":1234567890123456789}`) + + if err := migrateConnectorCredentials(context.Background(), db); err != nil { + t.Fatalf("migrate: %v", err) + } + + migrated := rawConnectorConfig(t, db, "c1") + if !strings.Contains(migrated, `"channel_id":1234567890123456789`) { + t.Fatalf("migrated config = %s, want channel_id unchanged", migrated) + } + var stored map[string]interface{} + if err := json.Unmarshal([]byte(migrated), &stored); err != nil { + t.Fatalf("migrated config %q is not JSON: %v", migrated, err) + } + credentials, _ := stored["credentials"].(string) + if !strings.HasPrefix(credentials, "enc:v1:") { + t.Fatalf("migrated config = %s, want enc:v1: credentials", migrated) + } + sealed, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(credentials, "enc:v1:")) + if err != nil { + t.Fatalf("migrated config = %s, want enc:v1: credentials", migrated) + } + key, err := common.ConnectorKey() + if err != nil { + t.Fatalf("key: %v", err) + } + block, err := aes.NewCipher(key) + if err != nil { + t.Fatalf("cipher: %v", err) + } + gcm, err := cipher.NewGCM(block) + if err != nil { + t.Fatalf("gcm: %v", err) + } + if len(sealed) < gcm.NonceSize()+gcm.Overhead() { + t.Fatalf("migrated config = %s, sealed value too short", migrated) + } + // Compare the plaintext bytes: DecryptConnectorCredentials decodes numbers to float64. + plaintext, err := gcm.Open(nil, sealed[:gcm.NonceSize()], sealed[gcm.NonceSize():], nil) + if err != nil { + t.Fatalf("decrypt migrated credentials: %v", err) + } + if string(plaintext) != `{"app_id":9007199254740993}` { + t.Fatalf("decrypted credentials = %s, want app_id 9007199254740993", plaintext) + } +} + +func TestMigrateConnectorCredentialsCoversEveryBatch(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsTestDB(t) + total := connectorCredentialsMigrationBatchSize + 1 + for i := 0; i < total; i++ { + insertRawConnector(t, db, fmt.Sprintf("c%04d", i), `{"credentials":{"api_token":"tok-123"}}`) + } + + if err := migrateConnectorCredentials(context.Background(), db); err != nil { + t.Fatalf("migrate: %v", err) + } + var encrypted int64 + if err := db.Table("connector").Where("config LIKE ?", `%"enc:v1:%`).Count(&encrypted).Error; err != nil { + t.Fatalf("count: %v", err) + } + if encrypted != int64(total) { + t.Fatalf("encrypted rows = %d, want %d", encrypted, total) + } +} + +func TestMigrateConnectorCredentialsWithoutKey(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, "") + db := setupConnectorCredentialsTestDB(t) + plain := `{"credentials": {"api_token": "tok-123"}}` + insertRawConnector(t, db, "plain", plain) + + if err := migrateConnectorCredentials(context.Background(), db); err != nil { + t.Fatalf("migrate: %v", err) + } + if got := rawConnectorConfig(t, db, "plain"); got != plain { + t.Fatalf("config = %q, want it untouched", got) + } +} + +func TestUpdateConnectorConfigIfUnchangedSkipsStaleRow(t *testing.T) { + db := setupConnectorCredentialsTestDB(t) + ctx := context.Background() + insertRawConnector(t, db, "c1", `{"credentials":"new"}`) + + updated, err := updateConnectorConfigIfUnchanged(ctx, db, "c1", `{"credentials":"old"}`, `{"credentials":"enc:v1:x"}`) + if err != nil || updated { + t.Fatalf("stale update: got (%v, %v), want (false, nil)", updated, err) + } + if got := rawConnectorConfig(t, db, "c1"); got != `{"credentials":"new"}` { + t.Fatalf("stale update overwrote the row: %q", got) + } + + updated, err = updateConnectorConfigIfUnchanged(ctx, db, "c1", `{"credentials":"new"}`, `{"credentials":"enc:v1:x"}`) + if err != nil || !updated { + t.Fatalf("current update: got (%v, %v), want (true, nil)", updated, err) + } + if got := rawConnectorConfig(t, db, "c1"); got != `{"credentials":"enc:v1:x"}` { + t.Fatalf("config = %q, want the new value", got) + } +} + +func TestConnectorWritesEncryptCredentials(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsTestDB(t) + ctx := context.Background() + connectorDAO := NewConnectorDAO() + assertEncrypted := func(step string) { + t.Helper() + raw := rawConnectorConfig(t, db, "c1") + if !strings.Contains(raw, `"credentials":"enc:v1:`) || strings.Contains(raw, "tok-") { + t.Fatalf("%s stored %s, want encrypted credentials", step, raw) + } + } + + if err := connectorDAO.Create(ctx, db, &entity.Connector{ + ID: "c1", TenantID: "tenant-1", Name: "c1", Source: "github", InputType: "poll", Status: "schedule", + Config: entity.ConnectorConfig{"credentials": map[string]interface{}{"api_token": "tok-123"}}, + }); err != nil { + t.Fatalf("create: %v", err) + } + assertEncrypted("create") + + updates := map[string]interface{}{"config": entity.ConnectorConfig{"credentials": map[string]interface{}{"api_token": "tok-456"}}} + if err := connectorDAO.UpdateByID(ctx, db, "c1", updates); err != nil { + t.Fatalf("update: %v", err) + } + assertEncrypted("map update") +} diff --git a/internal/dao/connector_test.go b/internal/dao/connector_test.go index 9f543766e7..53ee7bde30 100644 --- a/internal/dao/connector_test.go +++ b/internal/dao/connector_test.go @@ -57,8 +57,8 @@ func TestListByDatasetIDTxCollapsesDuplicateLinks(t *testing.T) { db := setupConnectorTestDB(t) connectors := []entity.Connector{ - {ID: "connector-a", TenantID: "tenant-1", Name: "A", Source: "s3", Status: "schedule", Config: entity.JSONMap{}}, - {ID: "connector-b", TenantID: "tenant-1", Name: "B", Source: "oss", Status: "schedule", Config: entity.JSONMap{}}, + {ID: "connector-a", TenantID: "tenant-1", Name: "A", Source: "s3", Status: "schedule", Config: entity.ConnectorConfig{}}, + {ID: "connector-b", TenantID: "tenant-1", Name: "B", Source: "oss", Status: "schedule", Config: entity.ConnectorConfig{}}, } if err := db.Create(&connectors).Error; err != nil { t.Fatalf("create connectors: %v", err) diff --git a/internal/dao/migration.go b/internal/dao/migration.go index b263df4f9f..526a9efc7f 100644 --- a/internal/dao/migration.go +++ b/internal/dao/migration.go @@ -81,6 +81,10 @@ func RunMigrations(ctx context.Context, db *gorm.DB) error { return fmt.Errorf("failed to migrate general chunker parser configs: %w", err) } + if err := migrateConnectorCredentials(ctx, db); err != nil { + return fmt.Errorf("failed to encrypt connector credentials: %w", err) + } + // Backfill the tenant model tables from the legacy tenant_llm table. if err := migrateModelData(ctx, db); err != nil { return fmt.Errorf("failed to migrate tenant model data: %w", err) diff --git a/internal/dao/sync_task_test.go b/internal/dao/sync_task_test.go index aa9052f0ca..6114608ba9 100644 --- a/internal/dao/sync_task_test.go +++ b/internal/dao/sync_task_test.go @@ -26,6 +26,7 @@ import ( "github.com/glebarez/sqlite" "gorm.io/gorm" + "ragflow/internal/common" "ragflow/internal/entity" ) @@ -230,3 +231,30 @@ func TestHandleTransientFailureUsesClassScopedRetryBudget(t *testing.T) { t.Fatalf("retry_count = %d, want 4", task.RetryCount) } } + +// TestGetTaskContextKeepsCredentialsEncrypted verifies that loading a task +// never needs the connector key, so claiming and failing a task still work +// after the key is lost. +func TestGetTaskContextKeepsCredentialsEncrypted(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, "") + db := setupConnectorCredentialsTestDB(t) + if err := db.AutoMigrate(&entity.Connector2Kb{}, &entity.Knowledgebase{}); err != nil { + t.Fatalf("migrate: %v", err) + } + insertRawConnector(t, db, "conn-1", `{"credentials":"enc:v1:abc","sync_deleted_files":true}`) + insertRunningSyncTask(t, db, "task-1", 0, 0, "") + if err := db.Create(&entity.Connector2Kb{ID: "link-1", ConnectorID: "conn-1", KbID: "kb-1", AutoParse: "1"}).Error; err != nil { + t.Fatalf("create link: %v", err) + } + if err := db.Create(&entity.Knowledgebase{ID: "kb-1", TenantID: "tenant-1", Name: "kb", EmbdID: "embd", Permission: "me", CreatedBy: "user-1", ParserID: "naive", ParserConfig: entity.JSONMap{}}).Error; err != nil { + t.Fatalf("create knowledgebase: %v", err) + } + + taskContext, err := NewSyncTaskDAO(db).GetTaskContext(context.Background(), "task-1") + if err != nil { + t.Fatalf("GetTaskContext: %v", err) + } + if taskContext.Connector.Config["credentials"] != "enc:v1:abc" || taskContext.Connector.Config["sync_deleted_files"] != true { + t.Fatalf("connector config = %#v, want the stored value", taskContext.Connector.Config) + } +} diff --git a/internal/entity/connector.go b/internal/entity/connector.go index f520df5227..27ee7ccee4 100644 --- a/internal/entity/connector.go +++ b/internal/entity/connector.go @@ -17,24 +17,27 @@ package entity import ( + "database/sql/driver" "encoding/json" "time" + + "ragflow/internal/common" ) // Connector connector model type Connector struct { - ID string `gorm:"column:id;primaryKey;size:32" json:"id"` - UserID string `gorm:"column:user_id;size:32;not null;default:'';index" json:"user_id"` - TenantID string `gorm:"column:tenant_id;size:32;not null;index" json:"tenant_id"` - Name string `gorm:"column:name;size:128;not null" json:"name"` - Source string `gorm:"column:source;size:128;not null;index" json:"source"` - InputType string `gorm:"column:input_type;size:128;not null;index" json:"input_type"` - Config JSONMap `gorm:"column:config;type:longtext;not null" json:"config"` - RefreshFreq int64 `gorm:"column:refresh_freq;default:0" json:"refresh_freq"` - PruneFreq int64 `gorm:"column:prune_freq;default:0" json:"prune_freq"` - TimeoutSecs int64 `gorm:"column:timeout_secs;default:3600" json:"timeout_secs"` - IndexingStart *time.Time `gorm:"column:indexing_start;index" json:"indexing_start,omitempty"` - Status string `gorm:"column:status;size:16;not null;default:schedule;index" json:"status"` + ID string `gorm:"column:id;primaryKey;size:32" json:"id"` + UserID string `gorm:"column:user_id;size:32;not null;default:'';index" json:"user_id"` + TenantID string `gorm:"column:tenant_id;size:32;not null;index" json:"tenant_id"` + Name string `gorm:"column:name;size:128;not null" json:"name"` + Source string `gorm:"column:source;size:128;not null;index" json:"source"` + InputType string `gorm:"column:input_type;size:128;not null;index" json:"input_type"` + Config ConnectorConfig `gorm:"column:config;type:longtext;not null" json:"config"` + RefreshFreq int64 `gorm:"column:refresh_freq;default:0" json:"refresh_freq"` + PruneFreq int64 `gorm:"column:prune_freq;default:0" json:"prune_freq"` + TimeoutSecs int64 `gorm:"column:timeout_secs;default:3600" json:"timeout_secs"` + IndexingStart *time.Time `gorm:"column:indexing_start;index" json:"indexing_start,omitempty"` + Status string `gorm:"column:status;size:16;not null;default:schedule;index" json:"status"` BaseModel } @@ -43,6 +46,40 @@ func (Connector) TableName() string { return "connector" } +// ConnectorConfig is the connector config column. Writes encrypt config["credentials"] when +// RAGFLOW_CONNECTOR_KEY is set. Reads keep it encrypted, so access checks and deletes work +// without the key; callers that need the credentials call common.DecryptConnectorCredentials. +type ConnectorConfig map[string]interface{} + +// Value implements driver.Valuer interface +func (c ConnectorConfig) Value() (driver.Value, error) { + if c == nil { + return nil, nil + } + config, err := common.EncryptConnectorCredentials(c) + if err != nil { + return nil, err + } + return json.Marshal(config) +} + +// Scan implements sql.Scanner interface +func (c *ConnectorConfig) Scan(value interface{}) error { + if value == nil { + *c = nil + return nil + } + b, ok := value.([]byte) + if !ok { + return json.Unmarshal([]byte(value.(string)), c) + } + return json.Unmarshal(b, c) +} + +func (c *ConnectorConfig) GormDataType() string { + return "longtext" +} + // MarshalJSON formats connector timestamps to match the Python API contract. func (c Connector) MarshalJSON() ([]byte, error) { type connectorJSON struct { @@ -69,7 +106,7 @@ func (c Connector) MarshalJSON() ([]byte, error) { Name: c.Name, Source: c.Source, InputType: c.InputType, - Config: c.Config, + Config: JSONMap(c.Config), RefreshFreq: c.RefreshFreq, PruneFreq: c.PruneFreq, TimeoutSecs: c.TimeoutSecs, diff --git a/internal/entity/connector_test.go b/internal/entity/connector_test.go new file mode 100644 index 0000000000..2a89810f47 --- /dev/null +++ b/internal/entity/connector_test.go @@ -0,0 +1,88 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed 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. +// + +package entity + +import ( + "encoding/json" + "strings" + "testing" + + "ragflow/internal/common" +) + +const testConnectorKey = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" + +func connectorConfigValue(t *testing.T, config ConnectorConfig) map[string]interface{} { + t.Helper() + value, err := config.Value() + if err != nil { + t.Fatalf("Value: %v", err) + } + raw, ok := value.([]byte) + if !ok { + t.Fatalf("Value returned %T, want []byte", value) + } + var stored map[string]interface{} + if err := json.Unmarshal(raw, &stored); err != nil { + t.Fatalf("stored value %q is not JSON: %v", raw, err) + } + return stored +} + +func TestConnectorConfigValueEncryptsCredentials(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + config := ConnectorConfig{"credentials": map[string]interface{}{"api_token": "tok-123"}, "wiki": "x"} + + stored := connectorConfigValue(t, config) + credentials, ok := stored["credentials"].(string) + if !ok || !strings.HasPrefix(credentials, "enc:v1:") { + t.Fatalf("stored credentials = %#v, want an enc:v1: string", stored["credentials"]) + } + if stored["wiki"] != "x" { + t.Fatalf("stored wiki = %#v, want x", stored["wiki"]) + } + if _, ok := config["credentials"].(map[string]interface{}); !ok { + t.Fatalf("Value mutated the config: %#v", config) + } +} + +func TestConnectorConfigValueWithoutKey(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, "") + stored := connectorConfigValue(t, ConnectorConfig{"credentials": map[string]interface{}{"api_token": "tok-123"}}) + credentials, ok := stored["credentials"].(map[string]interface{}) + if !ok || credentials["api_token"] != "tok-123" { + t.Fatalf("stored credentials = %#v, want the plaintext map", stored["credentials"]) + } +} + +func TestConnectorConfigValueFailsOnInvalidKey(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, "not-base64-!!") + if value, err := (ConnectorConfig{"credentials": "tok-123"}).Value(); err == nil { + t.Fatalf("Value = %q, want an error", value) + } +} + +func TestConnectorConfigScanDoesNotDecrypt(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + var config ConnectorConfig + if err := config.Scan([]byte(`{"credentials":"enc:v1:abc","wiki":"x"}`)); err != nil { + t.Fatalf("Scan: %v", err) + } + if config["credentials"] != "enc:v1:abc" || config["wiki"] != "x" { + t.Fatalf("scanned config = %#v, want the stored value unchanged", config) + } +} diff --git a/internal/handler/connector.go b/internal/handler/connector.go index 85a7b0f989..af064bcd26 100644 --- a/internal/handler/connector.go +++ b/internal/handler/connector.go @@ -273,12 +273,11 @@ func (h *ConnectorHandler) CreateConnector(c *gin.Context) { common.RespondError(c, common.CodeDataError, "config is required") return } - ctx := c.Request.Context() connector, err := h.connectorService.CreateConnector(ctx, &req) if err != nil { - if errors.Is(err, service.ErrConnectorInvalidFreq) || errors.Is(err, service.ErrConnectorInvalidName) { + if errors.Is(err, service.ErrConnectorInvalidFreq) || errors.Is(err, service.ErrConnectorInvalidName) || errors.Is(err, service.ErrConnectorEncryptedCredentials) { common.RespondError(c, common.CodeDataError, err.Error()) return } diff --git a/internal/handler/connector_test.go b/internal/handler/connector_test.go index a5488495c2..f9b3f20262 100644 --- a/internal/handler/connector_test.go +++ b/internal/handler/connector_test.go @@ -711,3 +711,23 @@ func TestConnectorHandlerListSyncLogs(t *testing.T) { }) } } + +func TestConnectorHandlerCreateConnectorRejectsEncryptedCredentials(t *testing.T) { + gin.SetMode(gin.TestMode) + h := &ConnectorHandler{connectorService: fakeConnectorService{err: service.ErrConnectorEncryptedCredentials}} + resp := httptest.NewRecorder() + c, _ := gin.CreateTestContext(resp) + c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/connectors", strings.NewReader(`{"name":"n","source":"github","config":{"credentials":"enc:v1:abc"}}`)) + c.Request.Header.Set("Content-Type", "application/json") + c.Set("user", &entity.User{ID: "user-1"}) + + h.CreateConnector(c) + + var body map[string]interface{} + if err := json.Unmarshal(resp.Body.Bytes(), &body); err != nil { + t.Fatalf("unmarshal response: %v", err) + } + if body["code"] != float64(common.CodeDataError) || body["message"] != service.ErrConnectorEncryptedCredentials.Error() { + t.Fatalf("body=%v, want CodeDataError and the encrypted credentials message", body) + } +} diff --git a/internal/handler/ingestion_messages_test.go b/internal/handler/ingestion_messages_test.go index 34c1a9b948..72fbee76e8 100644 --- a/internal/handler/ingestion_messages_test.go +++ b/internal/handler/ingestion_messages_test.go @@ -101,8 +101,8 @@ func TestDatasetsHandlerListIngestionLogsDatasourceLogType(t *testing.T) { insertIngestionMessagesHandlerKB(t, db, "kb-1", "user-1") connectors := []entity.Connector{ - {ID: "connector-1", TenantID: "user-1", Name: "configured", Source: "rss", InputType: "poll", Config: entity.JSONMap{}, Status: string(entity.TaskStatusDone)}, - {ID: "connector-2", TenantID: "user-1", Name: "unlinked", Source: "rss", InputType: "poll", Config: entity.JSONMap{}, Status: string(entity.TaskStatusDone)}, + {ID: "connector-1", TenantID: "user-1", Name: "configured", Source: "rss", InputType: "poll", Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusDone)}, + {ID: "connector-2", TenantID: "user-1", Name: "unlinked", Source: "rss", InputType: "poll", Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusDone)}, } if err := db.Create(&connectors).Error; err != nil { t.Fatalf("insert connectors: %v", err) diff --git a/internal/permission/database_source_test.go b/internal/permission/database_source_test.go index 820b96c814..b1326a6d6d 100644 --- a/internal/permission/database_source_test.go +++ b/internal/permission/database_source_test.go @@ -1030,9 +1030,9 @@ func TestDatabaseCheckerAuthorizesConnectorsAndSyncTasks(t *testing.T) { valid := string(entity.StatusValid) invited := entity.TenantMembershipStatusInvited connectors := []entity.Connector{ - {ID: "personal-connector", TenantID: "team-member", Name: "Personal", Source: "rest", InputType: "poll", Config: entity.JSONMap{}, Status: "5"}, - {ID: "team-connector", TenantID: "team-tenant", Name: "Team", Source: "rest", InputType: "poll", Config: entity.JSONMap{}, Status: "5"}, - {ID: "other-connector", TenantID: "other-tenant", Name: "Other", Source: "rest", InputType: "poll", Config: entity.JSONMap{}, Status: "5"}, + {ID: "personal-connector", TenantID: "team-member", Name: "Personal", Source: "rest", InputType: "poll", Config: entity.ConnectorConfig{}, Status: "5"}, + {ID: "team-connector", TenantID: "team-tenant", Name: "Team", Source: "rest", InputType: "poll", Config: entity.ConnectorConfig{}, Status: "5"}, + {ID: "other-connector", TenantID: "other-tenant", Name: "Other", Source: "rest", InputType: "poll", Config: entity.ConnectorConfig{}, Status: "5"}, } for i := range connectors { if err := db.Create(&connectors[i]).Error; err != nil { diff --git a/internal/server/config/environments.go b/internal/server/config/environments.go index 9842c85dce..549edde490 100644 --- a/internal/server/config/environments.go +++ b/internal/server/config/environments.go @@ -132,6 +132,11 @@ func (c *Config) GetEnvironments() error { c.environments.SecretKey = envVal } + // Fail at startup, not on the first connector write. + if _, err := common.ConnectorKey(); err != nil { + return err + } + if envVal := common.GetEnv(common.EnvEnableRegister); envVal != "" { if c.environments.EnableRegister == nil { c.environments.EnableRegister = new(bool) diff --git a/internal/server/config/environments_test.go b/internal/server/config/environments_test.go new file mode 100644 index 0000000000..adb96dff01 --- /dev/null +++ b/internal/server/config/environments_test.go @@ -0,0 +1,32 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed 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. +// + +package config + +import ( + "strings" + "testing" + + "ragflow/internal/common" +) + +func TestGetEnvironmentsRejectsInvalidConnectorKey(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, "c2hvcnQ=") + err := (&Config{}).GetEnvironments() + if err == nil || !strings.Contains(err.Error(), common.EnvRAGFlowConnectorKey) || strings.Contains(err.Error(), "c2hvcnQ=") { + t.Fatalf("GetEnvironments error = %v, want an error naming %s without its value", err, common.EnvRAGFlowConnectorKey) + } +} diff --git a/internal/service/connector.go b/internal/service/connector.go index f4ba3ffd1c..45c5fcf8e2 100644 --- a/internal/service/connector.go +++ b/internal/service/connector.go @@ -86,6 +86,8 @@ var ( ErrConnectorSourceNotImplemented = errors.New("connector source is not implemented") // ErrConnectorInternal is a generic, safe-to-expose internal failure. ErrConnectorInternal = errors.New("Internal server error") + // ErrConnectorEncryptedCredentials is returned when a request sends credentials in the stored "enc:" form. + ErrConnectorEncryptedCredentials = errors.New("config.credentials must be plaintext, not an \"enc:\" value") // ErrConnectorInvalidFreq is returned when a connector frequency is not // positive. Handlers map it to a client error; callers match with errors.Is. ErrConnectorInvalidFreq = errors.New("connector frequency must be positive") @@ -303,6 +305,9 @@ func (s *ConnectorService) CreateConnector(ctx context.Context, req *CreateConne if req.PruneFreq != nil && *req.PruneFreq <= 0 { return nil, fmt.Errorf("%w: prune_freq got %d", ErrConnectorInvalidFreq, *req.PruneFreq) } + if common.HasEncryptedConnectorCredentials(req.Config) { + return nil, ErrConnectorEncryptedCredentials + } refreshFreq := int64(defaultConnectorFreq) if req.RefreshFreq != nil { refreshFreq = *req.RefreshFreq @@ -325,7 +330,7 @@ func (s *ConnectorService) CreateConnector(ctx context.Context, req *CreateConne Name: req.Name, Source: req.Source, InputType: connectorInputTypePoll, - Config: req.Config, + Config: entity.ConnectorConfig(req.Config), RefreshFreq: refreshFreq, PruneFreq: pruneFreq, TimeoutSecs: timeoutSecs, @@ -341,7 +346,11 @@ func (s *ConnectorService) CreateConnector(ctx context.Context, req *CreateConne return nil, err } - return s.connectorDAO.GetByID(ctx, dao.DB, connector.ID) + created, err := s.connectorDAO.GetByID(ctx, dao.DB, connector.ID) + if err != nil { + return nil, err + } + return withDecryptedCredentials(created) } // GetConnector returns one connector when the user has read permission. @@ -362,6 +371,17 @@ func (s *ConnectorService) GetConnector(ctx context.Context, connectorID string) return nil, ErrConnectorInternal } + return withDecryptedCredentials(connector) +} + +// withDecryptedCredentials decrypts the credentials of a connector that goes +// back to the client. The web form sends this config back on update. +func withDecryptedCredentials(connector *entity.Connector) (*entity.Connector, error) { + config, err := common.DecryptConnectorCredentials(connector.Config) + if err != nil { + return nil, err + } + connector.Config = config return connector, nil } @@ -466,16 +486,19 @@ func testConnectorSettings(stored *entity.Connector, request entity.JSONMap) (st var config entity.JSONMap if stored != nil { source = strings.TrimSpace(stored.Source) - config = stored.Config + config = entity.JSONMap(stored.Config) } + fromRequest := false if request != nil { if value := strings.TrimSpace(stringConfigValue(request["source"])); value != "" { source = value } if nested, ok := request["config"]; ok { config = jsonMapValue(nested) + fromRequest = true } else if _, ok := request["source"]; !ok { config = request + fromRequest = true } } if source == "" { @@ -484,7 +507,18 @@ func testConnectorSettings(stored *entity.Connector, request entity.JSONMap) (st if config == nil { return "", nil, fmt.Errorf("connector configuration is missing") } - return source, config, nil + if fromRequest { + if common.HasEncryptedConnectorCredentials(config) { + return "", nil, &syncerconnector.ConnectorValidationError{Message: ErrConnectorEncryptedCredentials.Error()} + } + return source, config, nil + } + // Decrypt only the stored fallback, so testing new credentials works without the key. + decrypted, err := common.DecryptConnectorCredentials(config) + if err != nil { + return "", nil, err + } + return source, decrypted, nil } func jsonMapValue(value any) entity.JSONMap { @@ -1022,6 +1056,14 @@ func (s *ConnectorService) UpdateConnector(ctx context.Context, connectorID stri return nil, common.CodeServerError, err } + // The response returns the stored credentials. Fail before any write when they + // cannot be decrypted; a request with new credentials replaces them instead. + if req == nil || req.Config == nil { + if _, err = common.DecryptConnectorCredentials(connector.Config); err != nil { + return nil, common.CodeServerError, err + } + } + updates := map[string]interface{}{} if req != nil { if req.PruneFreq != nil { @@ -1037,7 +1079,11 @@ func (s *ConnectorService) UpdateConnector(ctx context.Context, connectorID stri updates["refresh_freq"] = *req.RefreshFreq } if req.Config != nil { - updates["config"] = req.Config + if common.HasEncryptedConnectorCredentials(req.Config) { + return nil, common.CodeDataError, ErrConnectorEncryptedCredentials + } + // GORM writes map update values as given, so the value itself must be the type that encrypts. + updates["config"] = entity.ConnectorConfig(req.Config) } if req.TimeoutSecs != nil { updates["timeout_secs"] = *req.TimeoutSecs @@ -1081,6 +1127,10 @@ func (s *ConnectorService) UpdateConnector(ctx context.Context, connectorID stri return nil, common.CodeServerError, err } + connector, err = withDecryptedCredentials(connector) + if err != nil { + return nil, common.CodeServerError, err + } return connector, common.CodeSuccess, nil } diff --git a/internal/service/connector_credentials_test.go b/internal/service/connector_credentials_test.go new file mode 100644 index 0000000000..fbffd47542 --- /dev/null +++ b/internal/service/connector_credentials_test.go @@ -0,0 +1,316 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed 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. +// + +package service + +import ( + "errors" + "strings" + "testing" + + "gorm.io/gorm" + + "ragflow/internal/common" + "ragflow/internal/dao" + "ragflow/internal/entity" + syncerconnector "ragflow/internal/syncer/connector" + connectormock "ragflow/internal/syncer/connector/mock" +) + +const testConnectorKey = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" + +func setupConnectorCredentialsServiceDB(t *testing.T) *gorm.DB { + t.Helper() + db := setupServiceTestDB(t) + pushServiceDB(t, db) + if err := db.AutoMigrate(&entity.Connector{}, &entity.SyncLogs{}); err != nil { + t.Fatalf("migrate connector tables: %v", err) + } + return db +} + +func insertConnectorWithToken(t *testing.T, db *gorm.DB, id, token string) { + t.Helper() + if err := db.Create(&entity.Connector{ + ID: id, + TenantID: "tenant-1", + Name: id, + Source: "mock", + InputType: "poll", + Config: entity.ConnectorConfig{"credentials": map[string]interface{}{"api_token": token}, "wiki": "x"}, + Status: "0", + }).Error; err != nil { + t.Fatalf("insert connector: %v", err) + } +} + +func rawConnectorConfigColumn(t *testing.T, db *gorm.DB, id string) string { + t.Helper() + var config string + if err := db.Raw("SELECT config FROM connector WHERE id = ?", id).Scan(&config).Error; err != nil { + t.Fatalf("read connector config: %v", err) + } + return config +} + +func connectorToken(config entity.ConnectorConfig) interface{} { + credentials, _ := config["credentials"].(map[string]interface{}) + return credentials["api_token"] +} + +func TestUpdateConnectorEncryptsCredentialsAtRest(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsServiceDB(t) + insertConnectorWithToken(t, db, "conn-1", "old-token") + + connector, code, err := NewConnectorService().UpdateConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1", &UpdateConnectorRequest{ + Config: entity.JSONMap{"credentials": map[string]interface{}{"api_token": "tok-123"}, "wiki": "y"}, + }) + if err != nil || code != common.CodeSuccess { + t.Fatalf("UpdateConnector: code=%v err=%v", code, err) + } + if connectorToken(connector.Config) != "tok-123" || connector.Config["wiki"] != "y" { + t.Fatalf("returned config = %#v, want plaintext credentials", connector.Config) + } + raw := rawConnectorConfigColumn(t, db, "conn-1") + if !strings.Contains(raw, `"credentials":"enc:v1:`) || strings.Contains(raw, "tok-123") { + t.Fatalf("stored config = %s, want encrypted credentials", raw) + } +} + +func TestCreateConnectorEncryptsCredentialsAtRest(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsServiceDB(t) + + connector, err := NewConnectorService().CreateConnector(serviceTestContext(t, "tenant-1", "tenant-1"), &CreateConnectorRequest{ + Name: "conn", + Source: "mock", + Config: entity.JSONMap{"credentials": map[string]interface{}{"api_token": "tok-123"}}, + }) + if err != nil { + t.Fatalf("CreateConnector: %v", err) + } + if connectorToken(connector.Config) != "tok-123" { + t.Fatalf("returned config = %#v, want plaintext credentials", connector.Config) + } + raw := rawConnectorConfigColumn(t, db, connector.ID) + if !strings.Contains(raw, `"credentials":"enc:v1:`) || strings.Contains(raw, "tok-123") { + t.Fatalf("stored config = %s, want encrypted credentials", raw) + } +} + +func TestGetConnectorDecryptsCredentials(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsServiceDB(t) + insertConnectorWithToken(t, db, "conn-1", "tok-123") + svc := NewConnectorService() + + connector, err := svc.GetConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1") + if err != nil { + t.Fatalf("GetConnector: %v", err) + } + if connectorToken(connector.Config) != "tok-123" { + t.Fatalf("returned config = %#v, want plaintext credentials", connector.Config) + } + + t.Setenv(common.EnvRAGFlowConnectorKey, "") + _, err = svc.GetConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1") + if err == nil || err.Error() != "connector credentials are encrypted but RAGFLOW_CONNECTOR_KEY is not set" { + t.Fatalf("GetConnector without key error = %v", err) + } +} + +func TestConnectorWithLostKeyCanBeUpdatedAndDeleted(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsServiceDB(t) + insertConnectorWithToken(t, db, "conn-1", "tok-123") + insertConnectorWithToken(t, db, "conn-2", "tok-456") + t.Setenv(common.EnvRAGFlowConnectorKey, "") + svc := NewConnectorService() + + connector, code, err := svc.UpdateConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1", &UpdateConnectorRequest{ + Config: entity.JSONMap{"credentials": map[string]interface{}{"api_token": "new-token"}}, + }) + if err != nil || code != common.CodeSuccess { + t.Fatalf("UpdateConnector: code=%v err=%v", code, err) + } + if connectorToken(connector.Config) != "new-token" { + t.Fatalf("returned config = %#v, want the new credentials", connector.Config) + } + + deleted, code, err := svc.DeleteConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-2") + if err != nil || code != common.CodeSuccess || !deleted { + t.Fatalf("DeleteConnector: deleted=%v code=%v err=%v", deleted, code, err) + } +} + +func TestUpdateConnectorWithLostKeyAndNoNewConfigChangesNothing(t *testing.T) { + refreshFreq := int64(30) + missingKey := "connector credentials are encrypted but RAGFLOW_CONNECTOR_KEY is not set" + for _, tc := range []struct { + name string + req *UpdateConnectorRequest + key string + err string + }{ + {name: "refresh_freq", req: &UpdateConnectorRequest{RefreshFreq: &refreshFreq}, err: missingKey}, + {name: "cancel", req: &UpdateConnectorRequest{Status: "CANCEL"}, err: missingKey}, + {name: "schedule", req: &UpdateConnectorRequest{Status: string(entity.TaskStatusSchedule)}, err: missingKey}, + {name: "reschedule", req: &UpdateConnectorRequest{Reschedule: true}, err: missingKey}, + // Bytes 1..32, not the key the row was written with. + {name: "wrong key", req: &UpdateConnectorRequest{RefreshFreq: &refreshFreq}, key: "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA=", err: "cannot decrypt connector credentials: wrong RAGFLOW_CONNECTOR_KEY or corrupted value"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsServiceDB(t) + if err := db.AutoMigrate(&entity.Connector2Kb{}, &entity.Knowledgebase{}); err != nil { + t.Fatalf("migrate kb tables: %v", err) + } + insertConnectorWithToken(t, db, "conn-1", "tok-123") + if err := db.Create(&entity.Knowledgebase{ID: "kb-1", TenantID: "tenant-1", Name: "kb-1", CreatedBy: "tenant-1", EmbdID: "embd"}).Error; err != nil { + t.Fatalf("insert kb: %v", err) + } + if err := db.Create(&entity.Connector2Kb{ID: "conn-1-kb-1", ConnectorID: "conn-1", KbID: "kb-1", AutoParse: "1"}).Error; err != nil { + t.Fatalf("insert connector2kb: %v", err) + } + if err := db.Create(&entity.SyncLogs{ + ID: "task-1", + ConnectorID: "conn-1", + KbID: "kb-1", + TaskType: dao.TaskTypeSync, + Status: string(entity.TaskStatusRunning), + }).Error; err != nil { + t.Fatalf("insert running task: %v", err) + } + t.Setenv(common.EnvRAGFlowConnectorKey, tc.key) + + _, code, err := NewConnectorService().UpdateConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1", tc.req) + if err == nil || err.Error() != tc.err || code != common.CodeServerError { + t.Fatalf("UpdateConnector: code=%v err=%v, want CodeServerError and %q", code, err, tc.err) + } + var connector entity.Connector + if err := db.First(&connector, "id = ?", "conn-1").Error; err != nil { + t.Fatalf("load connector: %v", err) + } + if connector.RefreshFreq != 0 || connector.Status != "0" { + t.Fatalf("connector refresh_freq/status = %d/%s, want 0/0", connector.RefreshFreq, connector.Status) + } + var task entity.SyncLogs + if err := db.First(&task, "id = ?", "task-1").Error; err != nil { + t.Fatalf("load task: %v", err) + } + if task.Status != string(entity.TaskStatusRunning) { + t.Fatalf("task status = %s, want running", task.Status) + } + var tasks int64 + if err := db.Model(&entity.SyncLogs{}).Count(&tasks).Error; err != nil { + t.Fatalf("count tasks: %v", err) + } + if tasks != 1 { + t.Fatalf("sync_logs rows = %d, want 1 (no task scheduled)", tasks) + } + }) + } +} + +func TestUpdateConnectorRejectsEncryptedCredentialsInput(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsServiceDB(t) + insertConnectorWithToken(t, db, "conn-1", "tok-123") + before := rawConnectorConfigColumn(t, db, "conn-1") + + _, code, err := NewConnectorService().UpdateConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1", &UpdateConnectorRequest{ + Config: entity.JSONMap{"credentials": "enc:v1:abc"}, + }) + if !errors.Is(err, ErrConnectorEncryptedCredentials) || code != common.CodeDataError { + t.Fatalf("UpdateConnector: code=%v err=%v, want CodeDataError and ErrConnectorEncryptedCredentials", code, err) + } + if after := rawConnectorConfigColumn(t, db, "conn-1"); after != before { + t.Fatalf("stored config changed: %s", after) + } +} + +func TestCreateConnectorRejectsEncryptedCredentialsInput(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsServiceDB(t) + + _, err := NewConnectorService().CreateConnector(serviceTestContext(t, "tenant-1", "tenant-1"), &CreateConnectorRequest{ + Name: "conn", + Source: "mock", + Config: entity.JSONMap{"credentials": "enc:v1:abc"}, + }) + if !errors.Is(err, ErrConnectorEncryptedCredentials) { + t.Fatalf("CreateConnector: err=%v, want ErrConnectorEncryptedCredentials", err) + } + var count int64 + if err := db.Model(&entity.Connector{}).Count(&count).Error; err != nil { + t.Fatalf("count connectors: %v", err) + } + if count != 0 { + t.Fatalf("connector rows = %d, want 0", count) + } +} + +func TestConnectorServiceTestConnectorDecryptsStoredCredentials(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, testConnectorKey) + db := setupConnectorCredentialsServiceDB(t) + insertConnectorWithToken(t, db, "conn-1", "tok-123") + + var capturedConfig map[string]any + registry := syncerconnector.NewRegistry() + registry.RegisterConfigFactory("mock", func(config map[string]any) (syncerconnector.Connector, error) { + capturedConfig = config + return &connectormock.Connector{}, nil + }) + svc := NewConnectorService() + svc.connectorRegistry = registry + + if err := svc.TestConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1", nil); err != nil { + t.Fatalf("TestConnector: %v", err) + } + if connectorToken(entity.ConnectorConfig(capturedConfig)) != "tok-123" { + t.Fatalf("config = %#v, want decrypted stored credentials", capturedConfig) + } + + t.Setenv(common.EnvRAGFlowConnectorKey, "") + request := entity.JSONMap{"source": "mock", "config": entity.JSONMap{"credentials": map[string]interface{}{"api_token": "new-token"}}} + if err := svc.TestConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1", request); err != nil { + t.Fatalf("TestConnector with request config and no key: %v", err) + } + err := svc.TestConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "conn-1", nil) + if err == nil || !strings.Contains(err.Error(), "RAGFLOW_CONNECTOR_KEY is not set") { + t.Fatalf("TestConnector with stored config and no key error = %v", err) + } +} + +func TestConnectorServiceTestConnectorRejectsEncryptedCredentialsInput(t *testing.T) { + setupConnectorCredentialsServiceDB(t) + registry := syncerconnector.NewRegistry() + registry.RegisterConfigFactory("mock", func(config map[string]any) (syncerconnector.Connector, error) { + return &connectormock.Connector{}, nil + }) + svc := NewConnectorService() + svc.connectorRegistry = registry + + err := svc.TestConnector(serviceTestContext(t, "tenant-1", "tenant-1"), "missing", entity.JSONMap{ + "source": "mock", + "config": entity.JSONMap{"credentials": "enc:v1:abc"}, + }) + var valErr *syncerconnector.ConnectorValidationError + if !errors.As(err, &valErr) || valErr.Message != ErrConnectorEncryptedCredentials.Error() { + t.Fatalf("error = %v, want *ConnectorValidationError for encrypted input", err) + } +} diff --git a/internal/service/connector_freq_test.go b/internal/service/connector_freq_test.go index 356f066b81..2fdef98482 100644 --- a/internal/service/connector_freq_test.go +++ b/internal/service/connector_freq_test.go @@ -22,7 +22,7 @@ func setupFreqValidationConnector(t *testing.T) { Name: "conn-freq", Source: "rss", InputType: "poll", - Config: entity.JSONMap{}, + Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusCancel), RefreshFreq: 5, PruneFreq: 5, diff --git a/internal/service/connector_logs_test.go b/internal/service/connector_logs_test.go index 47b9e47282..f545e5b433 100644 --- a/internal/service/connector_logs_test.go +++ b/internal/service/connector_logs_test.go @@ -39,7 +39,7 @@ func TestListLogsPaginationModes(t *testing.T) { Name: "conn-1", Source: "rss", InputType: "poll", - Config: entity.JSONMap{}, + Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusDone), RefreshFreq: 5, PruneFreq: 5, @@ -129,7 +129,7 @@ func TestListLogsStatusPriority(t *testing.T) { Name: "conn-1", Source: "rss", InputType: "poll", - Config: entity.JSONMap{}, + Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusSchedule), RefreshFreq: 5, PruneFreq: 5, diff --git a/internal/service/connector_syncer_test.go b/internal/service/connector_syncer_test.go index 25e41e7a37..24435f0fcf 100644 --- a/internal/service/connector_syncer_test.go +++ b/internal/service/connector_syncer_test.go @@ -58,7 +58,7 @@ func TestUpdateConnectorSchedulePublishesSyncerTask(t *testing.T) { Name: "conn-1", Source: "rss", InputType: "poll", - Config: entity.JSONMap{}, + Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusCancel), RefreshFreq: 0, PruneFreq: 0, @@ -125,7 +125,7 @@ func TestResumeFailedSyncSchedulesOriginalTaskFromCheckpoint(t *testing.T) { Name: "conn-1", Source: "gmail", InputType: "poll", - Config: entity.JSONMap{}, + Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusFail), RefreshFreq: 0, PruneFreq: 0, @@ -221,7 +221,7 @@ func TestRebuildConnectorDeletesOldSyncCheckpointsBeforePublishing(t *testing.T) Name: "conn-1", Source: "gmail", InputType: "poll", - Config: entity.JSONMap{}, + Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusFail), RefreshFreq: 0, PruneFreq: 0, @@ -302,7 +302,7 @@ func TestRebuildConnectorRejectsCrossTenantAndUnboundKB(t *testing.T) { } if err := db.Create(&entity.Connector{ ID: "conn-1", TenantID: "user-1", Name: "conn-1", Source: "gmail", - InputType: "poll", Config: entity.JSONMap{}, Status: string(entity.TaskStatusSchedule), + InputType: "poll", Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusSchedule), RefreshFreq: 0, PruneFreq: 0, TimeoutSecs: 60, }).Error; err != nil { t.Fatalf("insert connector: %v", err) @@ -357,7 +357,7 @@ func TestUpdateConnectorScheduleDoesNotDuplicateRunningTask(t *testing.T) { Name: "conn-1", Source: "rss", InputType: "poll", - Config: entity.JSONMap{}, + Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusRunning), RefreshFreq: 0, PruneFreq: 0, diff --git a/internal/service/connector_test_connection_test.go b/internal/service/connector_test_connection_test.go index 66cc9d4693..1d10a1dbc8 100644 --- a/internal/service/connector_test_connection_test.go +++ b/internal/service/connector_test_connection_test.go @@ -24,7 +24,7 @@ func TestConnectorServiceTestConnectorUsesRequestConfig(t *testing.T) { Name: "conn-1", Source: "mock", InputType: "poll", - Config: entity.JSONMap{"from": "stored"}, + Config: entity.ConnectorConfig{"from": "stored"}, Status: "0", }).Error; err != nil { t.Fatalf("insert connector: %v", err) @@ -123,7 +123,7 @@ func TestConnectorServiceTestConnectorRejectsUnauthorizedConnector(t *testing.T) Name: "conn-1", Source: "mock", InputType: "poll", - Config: entity.JSONMap{"ok": true}, + Config: entity.ConnectorConfig{"ok": true}, Status: "0", }).Error; err != nil { t.Fatalf("insert connector: %v", err) diff --git a/internal/service/dataset/ingestion_messages_test.go b/internal/service/dataset/ingestion_messages_test.go index f7ad1c6852..cbdc2012d0 100644 --- a/internal/service/dataset/ingestion_messages_test.go +++ b/internal/service/dataset/ingestion_messages_test.go @@ -121,9 +121,9 @@ func TestListSyncLogsReturnsOnlyCurrentlyLinkedDatasetConnectors(t *testing.T) { insertCompilationOwnerKB(t, "kb-2", "user-1") connectors := []entity.Connector{ - {ID: "connector-1", TenantID: "user-1", Name: "linked", Source: "rss", InputType: "poll", Config: entity.JSONMap{}, Status: string(entity.TaskStatusDone)}, - {ID: "connector-2", TenantID: "user-1", Name: "other", Source: "rss", InputType: "poll", Config: entity.JSONMap{}, Status: string(entity.TaskStatusDone)}, - {ID: "connector-3", TenantID: "user-2", Name: "inaccessible", Source: "rss", InputType: "poll", Config: entity.JSONMap{}, Status: string(entity.TaskStatusDone)}, + {ID: "connector-1", TenantID: "user-1", Name: "linked", Source: "rss", InputType: "poll", Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusDone)}, + {ID: "connector-2", TenantID: "user-1", Name: "other", Source: "rss", InputType: "poll", Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusDone)}, + {ID: "connector-3", TenantID: "user-2", Name: "inaccessible", Source: "rss", InputType: "poll", Config: entity.ConnectorConfig{}, Status: string(entity.TaskStatusDone)}, } if err := db.Create(&connectors).Error; err != nil { t.Fatalf("insert connectors: %v", err) diff --git a/internal/service/dataset/ingestion_summary_test.go b/internal/service/dataset/ingestion_summary_test.go index 14fa0a70ab..bf22997c64 100644 --- a/internal/service/dataset/ingestion_summary_test.go +++ b/internal/service/dataset/ingestion_summary_test.go @@ -24,10 +24,10 @@ func TestGetDownloadStatusAggregatesLinkedConnectorFiles(t *testing.T) { pushServiceDB(t, db) connectors := []entity.Connector{ - {ID: "connector-checkpoint", TenantID: "tenant-1", Name: "checkpoint", Source: "rss", InputType: "poll", Config: entity.JSONMap{}, Status: dao.SyncStatusSchedule}, - {ID: "connector-fallback", TenantID: "tenant-1", Name: "fallback", Source: "s3", InputType: "poll", Config: entity.JSONMap{}, Status: dao.SyncStatusSchedule}, - {ID: "connector-source-fail", TenantID: "tenant-1", Name: "source-fail", Source: "notion", InputType: "poll", Config: entity.JSONMap{}, Status: dao.SyncStatusSchedule}, - {ID: "connector-unlinked", TenantID: "tenant-1", Name: "unlinked", Source: "webdav", InputType: "poll", Config: entity.JSONMap{}, Status: dao.SyncStatusSchedule}, + {ID: "connector-checkpoint", TenantID: "tenant-1", Name: "checkpoint", Source: "rss", InputType: "poll", Config: entity.ConnectorConfig{}, Status: dao.SyncStatusSchedule}, + {ID: "connector-fallback", TenantID: "tenant-1", Name: "fallback", Source: "s3", InputType: "poll", Config: entity.ConnectorConfig{}, Status: dao.SyncStatusSchedule}, + {ID: "connector-source-fail", TenantID: "tenant-1", Name: "source-fail", Source: "notion", InputType: "poll", Config: entity.ConnectorConfig{}, Status: dao.SyncStatusSchedule}, + {ID: "connector-unlinked", TenantID: "tenant-1", Name: "unlinked", Source: "webdav", InputType: "poll", Config: entity.ConnectorConfig{}, Status: dao.SyncStatusSchedule}, } if err := db.Create(&connectors).Error; err != nil { t.Fatalf("create connectors: %v", err) diff --git a/internal/service/dataset/update_test.go b/internal/service/dataset/update_test.go index 7bf5a11718..489a090997 100644 --- a/internal/service/dataset/update_test.go +++ b/internal/service/dataset/update_test.go @@ -1116,7 +1116,7 @@ func insertDatasetUpdateConnector(t *testing.T, id, tenantID string) { Name: "Test Connector", Source: "google_drive", InputType: "oauth", - Config: entity.JSONMap{"sync_deleted_files": false}, + Config: entity.ConnectorConfig{"sync_deleted_files": false}, Status: string(entity.TaskStatusDone), } if err := dao.DB.Create(connector).Error; err != nil { diff --git a/internal/syncer/connector/builtin.go b/internal/syncer/connector/builtin.go index d6d76713ea..89b79eb308 100644 --- a/internal/syncer/connector/builtin.go +++ b/internal/syncer/connector/builtin.go @@ -20,6 +20,7 @@ import ( "context" "strings" + "ragflow/internal/common" "ragflow/internal/dao" ) @@ -86,6 +87,12 @@ func registerBuiltIn[T Connector](registry *Registry, source string, factory fun return factory(config) }) registry.Register(source, func(ctx context.Context, taskContext dao.SyncTaskContext) (Connector, error) { - return factory(map[string]any(taskContext.Connector.Config)) + // Decrypt here, not when the task is loaded: a missing key then fails + // this task through the normal retry path instead of blocking its claim. + config, err := common.DecryptConnectorCredentials(taskContext.Connector.Config) + if err != nil { + return nil, err + } + return factory(config) }) } diff --git a/internal/syncer/connector/feishu_wiki_test.go b/internal/syncer/connector/feishu_wiki_test.go index a2c2537391..3ec3765798 100644 --- a/internal/syncer/connector/feishu_wiki_test.go +++ b/internal/syncer/connector/feishu_wiki_test.go @@ -643,7 +643,7 @@ func TestFeishuWikiRegistryRegistration(t *testing.T) { // The task-context factory path must also resolve. taskConnector, err := registry.Open(context.Background(), dao.SyncTaskContext{ - Connector: entity.Connector{Source: "feishu_wiki", Config: entity.JSONMap(feishuWikiBaseConfig())}, + Connector: entity.Connector{Source: "feishu_wiki", Config: entity.ConnectorConfig(feishuWikiBaseConfig())}, }) if err != nil { t.Fatalf("Open feishu_wiki: %v", err) diff --git a/internal/syncer/connector/registry_test.go b/internal/syncer/connector/registry_test.go index 56c5659964..f52b2ba80d 100644 --- a/internal/syncer/connector/registry_test.go +++ b/internal/syncer/connector/registry_test.go @@ -7,6 +7,7 @@ import ( "io" "net/http" "net/http/httptest" + "ragflow/internal/common" "ragflow/internal/dao" "ragflow/internal/entity" "strconv" @@ -51,6 +52,34 @@ func TestRegistryOpenUsesTaskFactory(t *testing.T) { } } +func TestRegisterBuiltInsDecryptsStoredCredentials(t *testing.T) { + t.Setenv(common.EnvRAGFlowConnectorKey, "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=") + stored, err := common.EncryptConnectorCredentials(map[string]any{ + "credentials": map[string]any{"github_access_token": "tok-123"}, + "repository_owner": "ada", + }) + if err != nil { + t.Fatalf("encrypt: %v", err) + } + registry := NewRegistry() + RegisterBuiltIns(registry) + taskContext := dao.SyncTaskContext{Connector: entity.Connector{Source: "github", Config: entity.ConnectorConfig(stored)}} + + connector, err := registry.Open(context.Background(), taskContext) + if err != nil { + t.Fatalf("Open failed: %v", err) + } + github, ok := connector.(*GitHubConnector) + if !ok || github.token != "tok-123" || github.owner != "ada" { + t.Fatalf("connector = %#v, want a GitHub connector with the decrypted token", connector) + } + + t.Setenv(common.EnvRAGFlowConnectorKey, "") + if _, err := registry.Open(context.Background(), taskContext); err == nil || !strings.Contains(err.Error(), "RAGFLOW_CONNECTOR_KEY is not set") { + t.Fatalf("Open without key error = %v, want the missing key error", err) + } +} + func TestRegistryS3SourceHonorsBucketType(t *testing.T) { registry := NewRegistry() RegisterBuiltIns(registry) @@ -67,7 +96,7 @@ func TestRegistryS3SourceHonorsBucketType(t *testing.T) { t.Fatalf("OpenFromConfig failed: %v", err) } fromTask, err := registry.Open(context.Background(), dao.SyncTaskContext{ - Connector: entity.Connector{Source: "s3", Config: entity.JSONMap(config)}, + Connector: entity.Connector{Source: "s3", Config: entity.ConnectorConfig(config)}, }) if err != nil { t.Fatalf("Open failed: %v", err) diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index 541bbcd056..d0e7b2538f 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -270,7 +270,7 @@ func insertTaskContext(t *testing.T, db *gorm.DB, connectorID, kbID, taskID, tas Name: connectorID, Source: "mock", InputType: "poll", - Config: entity.JSONMap{"sync_deleted_files": true}, + Config: entity.ConnectorConfig{"sync_deleted_files": true}, // Keep frequencies positive: taskDelay treats freq <= 0 as // "no periodic reschedule" (fail-closed against hot-loop // republishing), so a zero fixture would silently suppress @@ -1421,6 +1421,63 @@ func TestTransientFailureFailsAfterThreeRetries(t *testing.T) { } } +// TestEncryptedCredentialsWithoutKeyFailAfterRetries verifies a lost +// RAGFLOW_CONNECTOR_KEY ends a built-in connector task as FAIL through the +// normal retry budget, and the stored error leaks no secret. +func TestEncryptedCredentialsWithoutKeyFailAfterRetries(t *testing.T) { + const key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" + t.Setenv(common.EnvRAGFlowConnectorKey, key) + db := setupSyncerDB(t) + insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) + if err := db.Model(&entity.Connector{}).Where("id = ?", "conn-1").Updates(map[string]any{ + "source": "github", + "config": entity.ConnectorConfig{"repository_owner": "octo", "credentials": map[string]any{"github_access_token": "tok-123"}}, + }).Error; err != nil { + t.Fatalf("set github connector: %v", err) + } + var connector entity.Connector + if err := db.First(&connector, "id = ?", "conn-1").Error; err != nil { + t.Fatalf("load connector: %v", err) + } + ciphertext, _ := connector.Config["credentials"].(string) + if !strings.HasPrefix(ciphertext, "enc:v1:") { + t.Fatalf("stored credentials = %#v, want an enc:v1: string", connector.Config["credentials"]) + } + t.Setenv(common.EnvRAGFlowConnectorKey, "") + + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) + registry := syncerconnector.NewRegistry() + syncerconnector.RegisterBuiltIns(registry) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, registry, &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) + + var task entity.SyncLogs + for attempt := 1; attempt <= 3; attempt++ { + claimed, err := taskService.Claim(t.Context(), "task-1") + if err != nil || !claimed { + t.Fatalf("attempt %d: claim = (%v, %v), want (true, nil)", attempt, claimed, err) + } + worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) + if err := db.First(&task, "id = ?", "task-1").Error; err != nil { + t.Fatalf("load task: %v", err) + } + if attempt < 3 && task.Status != dao.SyncStatusSchedule { + t.Fatalf("attempt %d: status = %s, want schedule for a retry", attempt, task.Status) + } + } + if task.Status != dao.SyncStatusFail { + t.Fatalf("status = %s, want fail", task.Status) + } + if !strings.Contains(task.ErrorMsg, "RAGFLOW_CONNECTOR_KEY is not set") { + t.Fatalf("error_msg = %q, want the missing key error", task.ErrorMsg) + } + for _, secret := range []string{strings.TrimPrefix(ciphertext, "enc:v1:"), key, "tok-123"} { + if strings.Contains(task.ErrorMsg, secret) { + t.Fatalf("error_msg = %q leaks %q", task.ErrorMsg, secret) + } + } +} + // TestErrorClassSwitchResetsRetryBudget verifies a task that switches failure // classes starts a fresh retry budget: failures accumulated under one class // must not consume the retries of a different class.