Merge branch 'main' into fix/#8834

Author: bujjibabukatta

.github/workflows/codespell.yml …lows/codespell.yml.action-blocked-by-asf.github/workflows/codespell.yml renamed to .github/workflows/codespell.yml.action-blocked-by-asf .github/workflows/codespell.yml renamed to .github/workflows/codespell.yml.action-blocked-by-asf

@@ -243,6 +243,7 @@ linters:
       - third_party$
       - builtin$
       - examples$
+      - mocks$
 formatters:
   enable:
     - gofmt

backend/.golangci.yaml

@@ -21,18 +21,23 @@ import (
 	"bytes"
 	"crypto/aes"
 	"crypto/cipher"
+	"crypto/rand"
 	"crypto/sha256"
 	"encoding/base64"
 	"fmt"
+	"io"
 
 	"github.com/apache/incubator-devlake/core/errors"
 	"github.com/apache/incubator-devlake/core/utils"
 )
 
 const EncodeKeyEnvStr = "ENCRYPTION_SECRET"
 
-// TODO: maybe move encryption/decryption into helper?
-// AES + Base64 encryption using ENCRYPTION_SECRET in .env as key
+// gcmNonceSize is the standard nonce size for AES-GCM.
+const gcmNonceSize = 12
+
+// Encrypt AES-GCM encrypts plaintext using ENCRYPTION_SECRET, then base64-encodes the result.
+// The output format is: base64(nonce || ciphertext || tag).
 func Encrypt(encryptionSecret, plainText string) (string, errors.Error) {
 	// add suffix to the data part
 	inputBytes := append([]byte(plainText), 123, 110, 100, 100, 116, 102, 125)
@@ -45,7 +50,8 @@ func Encrypt(encryptionSecret, plainText string) (string, errors.Error) {
 	return base64.StdEncoding.EncodeToString(output), nil
 }
 
-// Base64 + AES decryption using ENCRYPTION_SECRET in .env as key
+// Decrypt base64-decodes then AES-GCM decrypts ciphertext using ENCRYPTION_SECRET.
+// For backward compatibility, it also attempts AES-CBC decryption if the data looks like legacy format.
 func Decrypt(encryptionSecret, encryptedText string) (string, errors.Error) {
 	// when encryption key is not set
 	if encryptionSecret == "" {
@@ -98,41 +104,59 @@ func PKCS7UnPadding(origData []byte) []byte {
 	return origData[:(length - unpadding)]
 }
 
-// AesEncrypt AES encryption, CBC
+// AesEncrypt AES-256-GCM encrypts origData using key.
+// The returned bytes are: nonce (12 bytes) || ciphertext || tag.
 func AesEncrypt(origData, key []byte) ([]byte, errors.Error) {
-	// data alignment fill and encryption
 	sha256Key := sha256.Sum256(key)
-	key = sha256Key[:]
-	block, err := aes.NewCipher(key)
+	block, err := aes.NewCipher(sha256Key[:])
 	if err != nil {
 		return nil, errors.Convert(err)
 	}
-	// data alignment fill and encryption
-	blockSize := block.BlockSize()
-	origData = PKCS7Padding(origData, blockSize)
-	blockMode := cipher.NewCBCEncrypter(block, key[:blockSize])
-	crypted := make([]byte, len(origData))
-	blockMode.CryptBlocks(crypted, origData)
-	return crypted, nil
+	gcm, err := cipher.NewGCM(block)
+	if err != nil {
+		return nil, errors.Convert(err)
+	}
+	nonce := make([]byte, gcmNonceSize)
+	if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
+		return nil, errors.Convert(err)
+	}
+	ciphertext := gcm.Seal(nonce, nonce, origData, nil)
+	return ciphertext, nil
 }
 
-// AesDecrypt AES decryption
+// AesDecrypt decrypts crypted data using key.
+// It first tries AES-256-GCM (expects a 12-byte nonce prefix).
+// If that fails and the data length is a multiple of the AES block size (legacy CBC format),
+// it falls back to AES-256-CBC for backward compatibility.
 func AesDecrypt(crypted, key []byte) ([]byte, errors.Error) {
-	// Uniformly use sha256 to process as 32-bit Byte (256-bit bit)
 	sha256Key := sha256.Sum256(key)
-	key = sha256Key[:]
-	block, err := aes.NewCipher(key)
+	block, err := aes.NewCipher(sha256Key[:])
 	if err != nil {
 		return nil, errors.Convert(err)
 	}
-	// Get the block size and check whether the ciphertext length is legal
 	blockSize := block.BlockSize()
+
+	// Try GCM first if the data is long enough to contain a nonce.
+	if len(crypted) >= gcmNonceSize+blockSize {
+		gcm, err := cipher.NewGCM(block)
+		if err != nil {
+			return nil, errors.Convert(err)
+		}
+		nonce := crypted[:gcmNonceSize]
+		ciphertext := crypted[gcmNonceSize:]
+		plaintext, err := gcm.Open(nil, nonce, ciphertext, nil)
+		if err == nil {
+			return plaintext, nil
+		}
+		// GCM decryption failed; fall through to try legacy CBC.
+	}
+
+	// Legacy CBC fallback.
 	if len(crypted)%blockSize != 0 {
 		return nil, errors.Default.New(fmt.Sprintf("The length of the data to be decrypted is [%d], so cannot match the required block size [%d]", len(crypted), blockSize))
 	}
 
-	// Decrypt and unalign data
-	blockMode := cipher.NewCBCDecrypter(block, key[:blockSize])
+	blockMode := cipher.NewCBCDecrypter(block, sha256Key[:blockSize])
 	origData := make([]byte, len(crypted))
 	blockMode.CryptBlocks(origData, crypted)
 	origData = PKCS7UnPadding(origData)

backend/core/plugin/plugin_utils.go

@@ -18,8 +18,13 @@ limitations under the License.
 package plugin
 
 import (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/sha256"
+	"encoding/base64"
 	"testing"
 
+	"github.com/apache/incubator-devlake/core/errors"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -70,3 +75,67 @@ func TestEncode(t *testing.T) {
 		})
 	}
 }
+
+func TestGCMEncDec(t *testing.T) {
+	TestStr := "The string for testing"
+	encryptionSecret, _ := RandomEncryptionSecret()
+
+	// Encrypt with the new GCM format.
+	newCiphertext, err := Encrypt(encryptionSecret, TestStr)
+	assert.Empty(t, err)
+
+	// Decrypt the new format.
+	decodedNew, err := Decrypt(encryptionSecret, newCiphertext)
+	assert.Empty(t, err)
+	assert.Equal(t, TestStr, decodedNew)
+
+	// Ensure two encryptions of the same plaintext produce different ciphertexts (random nonce).
+	newCiphertext2, err := Encrypt(encryptionSecret, TestStr)
+	assert.Empty(t, err)
+	assert.NotEqual(t, newCiphertext, newCiphertext2)
+}
+
+// AesEncrypt AES encryption, CBC
+func oldAesEncrypt(origData, key []byte) ([]byte, errors.Error) {
+	// data alignment fill and encryption
+	sha256Key := sha256.Sum256(key)
+	key = sha256Key[:]
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, errors.Convert(err)
+	}
+	// data alignment fill and encryption
+	blockSize := block.BlockSize()
+	origData = PKCS7Padding(origData, blockSize)
+	blockMode := cipher.NewCBCEncrypter(block, key[:blockSize])
+	crypted := make([]byte, len(origData))
+	blockMode.CryptBlocks(crypted, origData)
+	return crypted, nil
+}
+
+func oldEncrypt(encryptionSecret, plainText string) (string, errors.Error) {
+	// add suffix to the data part
+	inputBytes := append([]byte(plainText), 123, 110, 100, 100, 116, 102, 125)
+	// perform encryption
+	output, err := oldAesEncrypt(inputBytes, []byte(encryptionSecret))
+	if err != nil {
+		return plainText, err
+	}
+	// Return the result after Base64 processing
+	return base64.StdEncoding.EncodeToString(output), nil
+}
+
+func TestBackwardCompatibility(t *testing.T) {
+	TestStr := "The string for testing"
+	encryptionSecret, _ := RandomEncryptionSecret()
+
+	// Encrypt with the new GCM format.
+	newCiphertext, err := oldEncrypt(encryptionSecret, TestStr)
+	assert.Empty(t, err)
+
+	// Decrypt the new format.
+	decodedNew, err := Decrypt(encryptionSecret, newCiphertext)
+	assert.Empty(t, err)
+	assert.Equal(t, TestStr, decodedNew)
+
+}

backend/core/plugin/plugin_utils_test.go

@@ -206,22 +206,22 @@ func TestMergeFromRequest_HandlesUsesApiToken(t *testing.T) {
 	// After merge, UsesApiToken should be updated
 	// This is a structural test - actual merge logic is in the connection.go MergeFromRequest method
 	assert.True(t, connection.UsesApiToken, "Initial value should be true")
-	
+
 	// If we were to apply the merge:
 	connection.UsesApiToken = newValues["usesApiToken"].(bool)
 	connection.Username = newValues["username"].(string)
-	
+
 	assert.False(t, connection.UsesApiToken, "After merge, should be false")
 	assert.Equal(t, "new_username", connection.Username)
 }
 
 func TestConnectionStatusCodes(t *testing.T) {
 	// Test expected status code handling
 	tests := []struct {
-		name           string
-		statusCode     int
-		expectedError  bool
-		errorType      string
+		name          string
+		statusCode    int
+		expectedError bool
+		errorType     string
 	}{
 		{
 			name:          "Success - 200 OK",

backend/plugins/bitbucket/api/connection_api_test.go

@@ -80,6 +80,15 @@ func listBitbucketWorkspaces(
 	if err != nil {
 		return
 	}
+	if res.StatusCode > 299 {
+		body, e := io.ReadAll(res.Body)
+		if e != nil {
+			err = errors.BadInput.Wrap(e, "failed to read response body")
+			return
+		}
+		err = errors.HttpStatus(res.StatusCode).New(string(body))
+		return
+	}
 
 	resBody := &models.WorkspaceResponse{}
 	err = api.UnmarshalResponse(res, resBody)

backend/plugins/bitbucket/api/remote_api.go

@@ -0,0 +1,82 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package e2e
+
+import (
+	"reflect"
+	"sort"
+	"testing"
+
+	"github.com/apache/incubator-devlake/helpers/e2ehelper"
+	"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+	"github.com/apache/incubator-devlake/plugins/circleci/impl"
+	"github.com/apache/incubator-devlake/plugins/circleci/models"
+	"github.com/apache/incubator-devlake/plugins/circleci/tasks"
+	"github.com/stretchr/testify/assert"
+)
+
+// TestCircleciUnfinishedJobsInputIterator is a regression test for
+// https://github.com/apache/devlake/issues/8907. The "collect unfinished job
+// details" collector builds its URL from "/v2/workflow/{{ .Input.Id }}/job" while
+// scanning rows into a models.CircleciJob. Its input query must therefore expose the
+// workflow id in the row's Id field; a bare "DISTINCT workflow_id" left Id empty and
+// produced "/v2/workflow//job" (HTTP 500). This test runs the production query
+// (tasks.UnfinishedJobsInputClauses) through the real iterator and asserts each
+// yielded row's Id is the workflow id, that results are DISTINCT, and that the
+// status/connection filters hold.
+func TestCircleciUnfinishedJobsInputIterator(t *testing.T) {
+	var circleci impl.Circleci
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+
+	const projectSlug = "github/test/repo"
+	dataflowTester.FlushTabler(&models.CircleciJob{})
+
+	seed := []models.CircleciJob{
+		{ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-1", ProjectSlug: projectSlug, Status: "on_hold"},
+		{ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-2", ProjectSlug: projectSlug, Status: "running"}, // same workflow -> DISTINCT
+		{ConnectionId: 1, WorkflowId: "wf-queued", Id: "job-3", ProjectSlug: projectSlug, Status: "queued"},
+		{ConnectionId: 1, WorkflowId: "wf-success", Id: "job-4", ProjectSlug: projectSlug, Status: "success"},   // terminal -> excluded
+		{ConnectionId: 2, WorkflowId: "wf-otherconn", Id: "job-5", ProjectSlug: projectSlug, Status: "on_hold"}, // other connection -> excluded
+	}
+	for i := range seed {
+		assert.Nil(t, dataflowTester.Dal.Create(&seed[i]))
+	}
+
+	cursor, err := dataflowTester.Dal.Cursor(tasks.UnfinishedJobsInputClauses(1, projectSlug)...)
+	assert.Nil(t, err)
+	iter, err := api.NewDalCursorIterator(dataflowTester.Dal, cursor, reflect.TypeOf(models.CircleciJob{}))
+	assert.Nil(t, err)
+	defer iter.Close()
+
+	var ids []string
+	for iter.HasNext() {
+		item, err := iter.Fetch()
+		assert.Nil(t, err)
+		job := item.(*models.CircleciJob)
+		ids = append(ids, job.Id)
+	}
+	sort.Strings(ids)
+
+	// Distinct workflow ids for connection 1's non-terminal jobs, with Id populated
+	// (the URL template reads .Input.Id). wf-success (terminal) and wf-otherconn
+	// (connection 2) are excluded.
+	assert.Equal(t, []string{"wf-onhold", "wf-queued"}, ids)
+	for _, id := range ids {
+		assert.NotEmpty(t, id, "Input.Id must be the workflow id, not empty (#8907)")
+	}
+}

backend/plugins/circleci/e2e/job_collector_test.go

@@ -41,6 +41,20 @@ var CollectJobsMeta = plugin.SubTaskMeta{
 	DomainTypes:      []string{plugin.DOMAIN_TYPE_CICD},
 }
 
+// UnfinishedJobsInputClauses returns the DAL clauses that select the workflows whose
+// jobs are still in a non-terminal status and therefore need their job details
+// recollected by the CollectJobs "unfinished details" collector.
+func UnfinishedJobsInputClauses(connectionId uint64, projectSlug string) []dal.Clause {
+	return []dal.Clause{
+		dal.Select("DISTINCT workflow_id AS id"), // #8907: alias to id so {{ .Input.Id }} resolves when scanned into CircleciJob
+		dal.From(&models.CircleciJob{}),
+		dal.Where(
+			"connection_id = ? AND project_slug = ? AND status IN ('running', 'not_running', 'queued', 'on_hold')",
+			connectionId, projectSlug,
+		),
+	}
+}
+
 func CollectJobs(taskCtx plugin.SubTaskContext) errors.Error {
 	rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_JOB_TABLE)
 	logger := taskCtx.GetLogger()
@@ -94,14 +108,8 @@ func CollectJobs(taskCtx plugin.SubTaskContext) errors.Error {
 				AfterResponse:  ignoreDeletedBuilds,
 			},
 			BuildInputIterator: func() (api.Iterator, errors.Error) {
-				clauses := []dal.Clause{
-					dal.Select("DISTINCT workflow_id"), // Only need to recollect jobs for a workflow once
-					dal.From(&models.CircleciJob{}),
-					dal.Where("connection_id = ? AND project_slug = ? AND status IN ('running', 'not_running', 'queued', 'on_hold')", data.Options.ConnectionId, data.Options.ProjectSlug),
-				}
-
 				db := taskCtx.GetDal()
-				cursor, err := db.Cursor(clauses...)
+				cursor, err := db.Cursor(UnfinishedJobsInputClauses(data.Options.ConnectionId, data.Options.ProjectSlug)...)
 				if err != nil {
 					return nil, err
 				}