fix(authZ): propagate new policies across replicas (#484)

Signed-off-by: Miguel Martinez Trivino <[email protected]>

Author: migmartri

app/controlplane/internal/authz/authz.go

@@ -17,15 +17,18 @@
 package authz
 
 import (
+	"context"
 	"errors"
 	"fmt"
 
 	_ "embed"
 
+	psqlwatcher "github.com/IguteChung/casbin-psql-watcher"
 	"github.com/casbin/casbin/v2"
 	"github.com/casbin/casbin/v2/model"
 	"github.com/casbin/casbin/v2/persist"
 	fileadapter "github.com/casbin/casbin/v2/persist/file-adapter"
+
 	entadapter "github.com/casbin/ent-adapter"
 	"github.com/chainloop-dev/chainloop/app/controlplane/internal/conf"
 )
@@ -123,6 +126,25 @@ func NewDatabaseEnforcer(c *conf.Data_Database) (*Enforcer, error) {
 		return nil, fmt.Errorf("failed to create enforcer: %w", err)
 	}
 
+	// watch for policy changes in database and update enforcer
+	w, err := psqlwatcher.NewWatcherWithConnString(context.Background(), c.Source, psqlwatcher.Option{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create watcher: %w", err)
+	}
+
+	if err = e.SetWatcher(w); err != nil {
+		return nil, fmt.Errorf("failed to set watcher: %w", err)
+	}
+
+	if err = w.SetUpdateCallback(func(string) {
+		// When there is a change in the policy, we load the in-memory policy for the current enforcer
+		if err := e.LoadPolicy(); err != nil {
+			fmt.Printf("failed to load policy: %v", err)
+		}
+	}); err != nil {
+		return nil, fmt.Errorf("failed to set update callback: %w", err)
+	}
+
 	return e, nil
 }
 

app/controlplane/internal/authz/authz_test.go

@@ -14,14 +14,18 @@
 // limitations under the License.
 
 // Authorization package
-package authz
+package authz_test
 
 import (
 	"fmt"
 	"io"
 	"os"
 	"testing"
+	"time"
 
+	"github.com/cenkalti/backoff/v4"
+	"github.com/chainloop-dev/chainloop/app/controlplane/internal/authz"
+	"github.com/chainloop-dev/chainloop/app/controlplane/internal/biz/testhelpers"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -30,8 +34,8 @@ import (
 func TestAddPolicies(t *testing.T) {
 	testcases := []struct {
 		name               string
-		subject            *SubjectAPIToken
-		policies           []*Policy
+		subject            *authz.SubjectAPIToken
+		policies           []*authz.Policy
 		wantErr            bool
 		wantNumberPolicies int
 	}{
@@ -41,38 +45,38 @@ func TestAddPolicies(t *testing.T) {
 		},
 		{
 			name: "no subject",
-			policies: []*Policy{
-				PolicyWorkflowContractList,
+			policies: []*authz.Policy{
+				authz.PolicyWorkflowContractList,
 			},
 			wantErr: true,
 		},
 		{
 			name:    "no policies",
-			subject: &SubjectAPIToken{ID: uuid.NewString()},
+			subject: &authz.SubjectAPIToken{ID: uuid.NewString()},
 			wantErr: true,
 		},
 		{
 			name:    "adds two policies",
-			subject: &SubjectAPIToken{ID: uuid.NewString()},
-			policies: []*Policy{
-				PolicyWorkflowContractList,
-				PolicyReferrerRead,
+			subject: &authz.SubjectAPIToken{ID: uuid.NewString()},
+			policies: []*authz.Policy{
+				authz.PolicyWorkflowContractList,
+				authz.PolicyReferrerRead,
 			},
 			wantNumberPolicies: 2,
 		},
 		{
 			name: "handles duplicated policies",
-			subject: &SubjectAPIToken{
+			subject: &authz.SubjectAPIToken{
 				ID: uuid.NewString(),
 			},
-			policies: []*Policy{
-				PolicyWorkflowContractList,
-				PolicyWorkflowContractRead,
-				PolicyWorkflowContractUpdate,
-				PolicyWorkflowContractList,
-				PolicyArtifactDownload,
-				PolicyWorkflowContractUpdate,
-				PolicyArtifactDownload,
+			policies: []*authz.Policy{
+				authz.PolicyWorkflowContractList,
+				authz.PolicyWorkflowContractRead,
+				authz.PolicyWorkflowContractUpdate,
+				authz.PolicyWorkflowContractList,
+				authz.PolicyArtifactDownload,
+				authz.PolicyWorkflowContractUpdate,
+				authz.PolicyArtifactDownload,
 			},
 			wantNumberPolicies: 4,
 		},
@@ -103,22 +107,22 @@ func TestAddPolicies(t *testing.T) {
 }
 
 func TestAddPoliciesDuplication(t *testing.T) {
-	want := []*Policy{
-		PolicyWorkflowContractList,
-		PolicyWorkflowContractRead,
+	want := []*authz.Policy{
+		authz.PolicyWorkflowContractList,
+		authz.PolicyWorkflowContractRead,
 	}
 
 	enforcer, closer := testEnforcer(t)
 	defer closer.Close()
-	sub := &SubjectAPIToken{ID: uuid.NewString()}
+	sub := &authz.SubjectAPIToken{ID: uuid.NewString()}
 
 	err := enforcer.AddPolicies(sub, want...)
 	require.NoError(t, err)
 	got := enforcer.GetFilteredPolicy(0, sub.String())
 	assert.Len(t, got, 2)
 
 	// Update the list of policies we want to add by appending an extra one
-	want = append(want, PolicyWorkflowContractUpdate)
+	want = append(want, authz.PolicyWorkflowContractUpdate)
 	// AddPolicies only add the policies that are not already present preventing duplication
 	err = enforcer.AddPolicies(sub, want...)
 	assert.NoError(t, err)
@@ -127,15 +131,15 @@ func TestAddPoliciesDuplication(t *testing.T) {
 }
 
 func TestClearPolicies(t *testing.T) {
-	want := []*Policy{
-		PolicyWorkflowContractList,
-		PolicyWorkflowContractRead,
+	want := []*authz.Policy{
+		authz.PolicyWorkflowContractList,
+		authz.PolicyWorkflowContractRead,
 	}
 
 	enforcer, closer := testEnforcer(t)
 	defer closer.Close()
-	sub := &SubjectAPIToken{ID: uuid.NewString()}
-	sub2 := &SubjectAPIToken{ID: uuid.NewString()}
+	sub := &authz.SubjectAPIToken{ID: uuid.NewString()}
+	sub2 := &authz.SubjectAPIToken{ID: uuid.NewString()}
 
 	// Create policies for two different subjects
 	err := enforcer.AddPolicies(sub, want...)
@@ -160,13 +164,68 @@ func TestClearPolicies(t *testing.T) {
 	assert.Len(t, got, 2)
 }
 
-func testEnforcer(t *testing.T) (*Enforcer, io.Closer) {
+func testEnforcer(t *testing.T) (*authz.Enforcer, io.Closer) {
 	f, err := os.CreateTemp(t.TempDir(), "policy*.csv")
 	if err != nil {
 		require.FailNow(t, err.Error())
 	}
 
-	enforcer, err := NewFiletypeEnforcer(f.Name())
+	enforcer, err := authz.NewFiletypeEnforcer(f.Name())
 	require.NoError(t, err)
 	return enforcer, f
 }
+
+func TestMultiReplicaPropagation(t *testing.T) {
+	// Create two enforcers that share the same database
+	db := testhelpers.NewTestDatabase(t)
+	defer db.Close(t)
+
+	enforcerA, err := authz.NewDatabaseEnforcer(testhelpers.NewConfData(db, t).Database)
+	require.NoError(t, err)
+	enforcerB, err := authz.NewDatabaseEnforcer(testhelpers.NewConfData(db, t).Database)
+	require.NoError(t, err)
+
+	// Subject and policies to add
+	sub := &authz.SubjectAPIToken{ID: uuid.NewString()}
+	want := []*authz.Policy{authz.PolicyWorkflowContractList, authz.PolicyWorkflowContractRead}
+
+	// Create policies in one enforcer
+	err = enforcerA.AddPolicies(sub, want...)
+	require.NoError(t, err)
+
+	// Make sure it propagates to the other one
+	got := enforcerA.GetFilteredPolicy(0, sub.String())
+	assert.Len(t, got, 2)
+
+	// it might take a bit for the policies to propagate to the other enforcer
+	err = fnWithRetry(func() error {
+		got = enforcerB.GetFilteredPolicy(0, sub.String())
+		if len(got) == 2 {
+			return nil
+		}
+		return fmt.Errorf("policies not propagated yet")
+	})
+	require.NoError(t, err)
+	assert.Len(t, got, 2)
+
+	// Then delete them from the second one and check propagation again
+	require.NoError(t, enforcerB.ClearPolicies(sub))
+	assert.Len(t, enforcerB.GetFilteredPolicy(0, sub.String()), 0)
+
+	// Make sure it propagates to the other one
+	err = fnWithRetry(func() error {
+		got = enforcerA.GetFilteredPolicy(0, sub.String())
+		if len(got) == 0 {
+			return nil
+		}
+
+		return fmt.Errorf("policies not propagated yet")
+	})
+	require.NoError(t, err)
+	assert.Len(t, enforcerA.GetFilteredPolicy(0, sub.String()), 0)
+}
+
+func fnWithRetry(f func() error) error {
+	// Max 1 seconds
+	return backoff.Retry(f, backoff.WithMaxRetries(backoff.NewConstantBackOff(100*time.Millisecond), 10))
+}

app/controlplane/internal/biz/testhelpers/database.go

@@ -179,7 +179,7 @@ func (db *TestDatabase) ConnectionString(t *testing.T) string {
 	return fmt.Sprintf("postgres://postgres:[email protected]:%d/postgres", db.Port(t))
 }
 
-func newConfData(db *TestDatabase, t *testing.T) *conf.Data {
+func NewConfData(db *TestDatabase, t *testing.T) *conf.Data {
 	return &conf.Data{Database: &conf.Data_Database{Driver: "pgx", Source: db.ConnectionString(t)}}
 }
 

app/controlplane/internal/biz/testhelpers/wire.go

@@ -44,7 +44,7 @@ func WireTestData(*TestDatabase, *testing.T, log.Logger, credentials.ReaderWrite
 			wire.Value(&conf.ReferrerSharedIndex{}),
 			wire.Struct(new(TestingUseCases), "*"),
 			wire.Struct(new(TestingRepos), "*"),
-			newConfData,
+			NewConfData,
 			authz.NewDatabaseEnforcer,
 			wire.FieldsOf(new(*conf.Data), "Database"),
 		),

app/controlplane/internal/biz/testhelpers/wire_gen.go

@@ -116,6 +116,7 @@ require (
 	github.com/jackc/pgproto3/v2 v2.3.2 // indirect
 	github.com/jackc/pgtype v1.14.0 // indirect
 	github.com/jackc/pgx/v4 v4.18.1 // indirect
+	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/kevinburke/ssh_config v1.2.0 // indirect
@@ -151,6 +152,7 @@ require (
 	cloud.google.com/go/iam v1.1.4 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/keyvault/azsecrets v0.12.0
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
+	github.com/IguteChung/casbin-psql-watcher v1.0.0
 	github.com/Microsoft/go-winio v0.6.1 // indirect
 	github.com/ProtonMail/go-crypto v0.0.0-20230923063757-afb1ddc0824c // indirect
 	github.com/ThalesIgnite/crypto11 v1.2.5 // indirect

go.mod

@@ -115,6 +115,8 @@ github.com/CycloneDX/cyclonedx-go v0.8.0/go.mod h1:K2bA+324+Og0X84fA8HhN2X066K7B
 github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60=
 github.com/DATA-DOG/go-sqlmock v1.5.0/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
 github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
+github.com/IguteChung/casbin-psql-watcher v1.0.0 h1:GO5RvdHq5WZfuKt03Frk4/SvYvikMI5V1zjSt1P1suM=
+github.com/IguteChung/casbin-psql-watcher v1.0.0/go.mod h1:mwQYBxiYGO05U8vogls8gf3KnOmdlc7GOLX7tS/V2TU=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
 github.com/Masterminds/semver/v3 v3.1.1 h1:hLg3sBzpNErnxhQtUy/mmLR2I9foDujNK030IGemrRc=
 github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=
@@ -855,6 +857,8 @@ github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0f
 github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.3.0/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
+github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
+github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
 github.com/jedib0t/go-pretty/v6 v6.4.7 h1:lwiTJr1DEkAgzljsUsORmWsVn5MQjt1BPJdPCtJ6KXE=