Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503package rbac_test
import ( "database/sql" "testing"
"tangled.org/core/rbac"
adapter "github.com/Blank-Xu/sql-adapter" "github.com/casbin/casbin/v2" "github.com/casbin/casbin/v2/model" _ "github.com/mattn/go-sqlite3" "github.com/stretchr/testify/assert")
func setup(t *testing.T) *rbac.Enforcer { db, err := sql.Open("sqlite3", ":memory:?_foreign_keys=1") assert.NoError(t, err)
a, err := adapter.NewAdapter(db, "sqlite3", "acl") assert.NoError(t, err)
m, err := model.NewModelFromString(rbac.Model) assert.NoError(t, err)
e, err := casbin.NewSyncedEnforcer(m, a) assert.NoError(t, err)
e.EnableAutoSave(false)
return &rbac.Enforcer{E: e}}
func TestAddKnotAndRoles(t *testing.T) { e := setup(t)
err := e.AddKnot("example.com") assert.NoError(t, err)
err = e.AddKnotOwner("example.com", "did:plc:foo") assert.NoError(t, err)
isOwner, err := e.IsKnotOwner("did:plc:foo", "example.com") assert.NoError(t, err) assert.True(t, isOwner)
isMember, err := e.IsKnotMember("did:plc:foo", "example.com") assert.NoError(t, err) assert.True(t, isMember)}
func TestAddMember(t *testing.T) { e := setup(t)
err := e.AddKnot("example.com") assert.NoError(t, err)
err = e.AddKnotOwner("example.com", "did:plc:foo") assert.NoError(t, err)
err = e.AddKnotMember("example.com", "did:plc:bar") assert.NoError(t, err)
isMember, err := e.IsKnotMember("did:plc:foo", "example.com") assert.NoError(t, err) assert.True(t, isMember)
isMember, err = e.IsKnotMember("did:plc:bar", "example.com") assert.NoError(t, err) assert.True(t, isMember)
isOwner, err := e.IsKnotOwner("did:plc:foo", "example.com") assert.NoError(t, err) assert.True(t, isOwner)
// negated check here isOwner, err = e.IsKnotOwner("did:plc:bar", "example.com") assert.NoError(t, err) assert.False(t, isOwner)}
func TestAddRepoPermissions(t *testing.T) { e := setup(t)
knot := "example.com"
fooUser := "did:plc:foo" fooRepo := "did:plc:foo/my-repo"
barUser := "did:plc:bar" barRepo := "did:plc:bar/my-repo"
_ = e.AddKnot(knot) _ = e.AddKnotMember(knot, fooUser) _ = e.AddKnotMember(knot, barUser)
err := e.AddRepo(fooUser, knot, fooRepo) assert.NoError(t, err)
err = e.AddRepo(barUser, knot, barRepo) assert.NoError(t, err)
canPush, err := e.IsPushAllowed(fooUser, knot, fooRepo) assert.NoError(t, err) assert.True(t, canPush)
canPush, err = e.IsPushAllowed(barUser, knot, barRepo) assert.NoError(t, err) assert.True(t, canPush)
// negated canPush, err = e.IsPushAllowed(barUser, knot, fooRepo) assert.NoError(t, err) assert.False(t, canPush)
canDelete, err := e.E.Enforce(fooUser, knot, fooRepo, "repo:delete") assert.NoError(t, err) assert.True(t, canDelete)
// negated canDelete, err = e.E.Enforce(barUser, knot, fooRepo, "repo:delete") assert.NoError(t, err) assert.False(t, canDelete)}
func TestCollaboratorPermissions(t *testing.T) { e := setup(t)
knot := "example.com" repo := "did:plc:foo/my-repo" owner := "did:plc:foo" collaborator := "did:plc:bar"
_ = e.AddKnot(knot) _ = e.AddRepo(owner, knot, repo)
err := e.AddCollaborator(collaborator, knot, repo) assert.NoError(t, err)
// all collaborator permissions granted perms := e.GetPermissionsInRepo(collaborator, knot, repo) assert.ElementsMatch(t, []string{ "repo:settings", "repo:push", "repo:collaborator", }, perms)
err = e.RemoveCollaborator(collaborator, knot, repo) assert.NoError(t, err)
// all permissions removed perms = e.GetPermissionsInRepo(collaborator, knot, repo) assert.ElementsMatch(t, []string{}, perms)}
func TestGetCollaboratorsByRepo(t *testing.T) { e := setup(t)
knot := "example.com" repo := "did:plc:foo/my-repo" otherRepo := "did:plc:foo/other-repo" owner := "did:plc:foo" collaborator1 := "did:plc:bar" collaborator2 := "did:plc:baz"
_ = e.AddKnot(knot) _ = e.AddRepo(owner, knot, repo) _ = e.AddRepo(owner, knot, otherRepo)
err := e.AddCollaborator(collaborator1, knot, repo) assert.NoError(t, err)
err = e.AddCollaborator(collaborator2, knot, repo) assert.NoError(t, err)
byRepo, err := e.GetCollaboratorsByRepo(knot) assert.NoError(t, err) assert.ElementsMatch(t, []string{ "did:plc:bar", // collaborator1 "did:plc:baz", // collaborator2 }, byRepo[repo]) assert.NotContains(t, byRepo[repo], owner, "owner does not hold repo:collaborator and must not be listed") assert.Empty(t, byRepo[otherRepo], "a repo without collaborators must not appear")}
func TestGetPermissionsInRepo(t *testing.T) { e := setup(t)
user := "did:plc:foo" knot := "example.com" repo := "did:plc:foo/my-repo"
_ = e.AddKnot(knot) _ = e.AddRepo(user, knot, repo)
perms := e.GetPermissionsInRepo(user, knot, repo) assert.ElementsMatch(t, []string{ "repo:settings", "repo:push", "repo:owner", "repo:invite", "repo:delete", }, perms)}
func TestInvalidRepoFormat(t *testing.T) { e := setup(t)
err := e.AddRepo("did:plc:foo", "example.com", "not-valid-format") assert.Error(t, err)}
func TestGetKnotssForUser(t *testing.T) { e := setup(t) _ = e.AddKnot("example.com") _ = e.AddKnotOwner("example.com", "did:plc:foo") _ = e.AddKnotMember("example.com", "did:plc:bar")
knots1, _ := e.GetKnotsForUser("did:plc:foo") assert.Contains(t, knots1, "example.com")
knots2, _ := e.GetKnotsForUser("did:plc:bar") assert.Contains(t, knots2, "example.com")}
func TestGetKnotUsersByRole(t *testing.T) { e := setup(t) _ = e.AddKnot("example.com") _ = e.AddKnotMember("example.com", "did:plc:foo") _ = e.AddKnotOwner("example.com", "did:plc:bar")
members, _ := e.GetKnotUsersByRole("server:member", "example.com") assert.Contains(t, members, "did:plc:foo") assert.Contains(t, members, "did:plc:bar") // due to inheritance}
func TestGetSpindleUsersByRole(t *testing.T) { e := setup(t) _ = e.AddSpindle("example.com") _ = e.AddSpindleMember("example.com", "did:plc:foo") _ = e.AddSpindleOwner("example.com", "did:plc:bar")
members, _ := e.GetSpindleUsersByRole("server:member", "example.com") assert.Contains(t, members, "did:plc:foo") assert.Contains(t, members, "did:plc:bar") // due to inheritance}
func TestEmptyUserPermissions(t *testing.T) { e := setup(t) allowed, _ := e.IsPushAllowed("did:plc:nobody", "unknown.com", "did:plc:nobody/repo") assert.False(t, allowed)}
func TestDuplicatePolicyAddition(t *testing.T) { e := setup(t) _ = e.AddKnot("example.com") _ = e.AddRepo("did:plc:foo", "example.com", "did:plc:foo/repo")
// add again err := e.AddRepo("did:plc:foo", "example.com", "did:plc:foo/repo") assert.NoError(t, err) // should not fail, but won't duplicate}
func TestRemoveRepo(t *testing.T) { e := setup(t) repo := "did:plc:foo/repo" _ = e.AddKnot("example.com") _ = e.AddRepo("did:plc:foo", "example.com", repo)
allowed, _ := e.IsSettingsAllowed("did:plc:foo", "example.com", repo) assert.True(t, allowed)
_ = e.RemoveRepo("did:plc:foo", "example.com", repo)
allowed, _ = e.IsSettingsAllowed("did:plc:foo", "example.com", repo) assert.False(t, allowed)}
func TestAddKnotAndSpindle(t *testing.T) { e := setup(t)
err := e.AddKnot("k.com") assert.NoError(t, err)
err = e.AddSpindle("s.com") assert.NoError(t, err)
err = e.AddKnotOwner("k.com", "did:plc:foo") assert.NoError(t, err)
err = e.AddSpindleOwner("s.com", "did:plc:foo") assert.NoError(t, err)
knots, err := e.GetKnotsForUser("did:plc:foo") assert.NoError(t, err) assert.ElementsMatch(t, []string{ "k.com", }, knots)
spindles, err := e.GetSpindlesForUser("did:plc:foo") assert.NoError(t, err) assert.ElementsMatch(t, []string{ "s.com", }, spindles)}
func TestAddSpindleAndRoles(t *testing.T) { e := setup(t)
err := e.AddSpindle("s.com") assert.NoError(t, err)
err = e.AddSpindleOwner("s.com", "did:plc:foo") assert.NoError(t, err)
ok, err := e.IsSpindleOwner("did:plc:foo", "s.com") assert.NoError(t, err) assert.True(t, ok)
ok, err = e.IsSpindleMember("did:plc:foo", "s.com") assert.NoError(t, err) assert.True(t, ok)}
func TestRemoveKnotOwner(t *testing.T) { e := setup(t)
err := e.AddKnot("k.com") assert.NoError(t, err)
err = e.AddKnotOwner("k.com", "did:plc:foo") assert.NoError(t, err)
knots, err := e.GetKnotsForUser("did:plc:foo") assert.NoError(t, err) assert.ElementsMatch(t, []string{ "k.com", }, knots)
err = e.RemoveKnotOwner("k.com", "did:plc:foo") assert.NoError(t, err)
knots, err = e.GetKnotsForUser("did:plc:foo") assert.NoError(t, err) assert.Empty(t, knots)}
func TestRemoveKnotMember(t *testing.T) { e := setup(t)
err := e.AddKnot("k.com") assert.NoError(t, err)
err = e.AddKnotOwner("k.com", "did:plc:foo") assert.NoError(t, err)
err = e.AddKnotMember("k.com", "did:plc:bar") assert.NoError(t, err)
knots, err := e.GetKnotsForUser("did:plc:bar") assert.NoError(t, err) assert.ElementsMatch(t, []string{ "k.com", }, knots)
err = e.RemoveKnotMember("k.com", "did:plc:bar") assert.NoError(t, err)
knots, err = e.GetKnotsForUser("did:plc:bar") assert.NoError(t, err) assert.Empty(t, knots)}
func TestRemoveKnotRemovesRepoPolicies(t *testing.T) { e := setup(t)
knot := "kelp.example" owner := "did:plc:akshay" collaborator := "did:plc:boltless" repo := "did:plc:akshay/anemone"
assert.NoError(t, e.AddKnot(knot)) assert.NoError(t, e.AddKnotOwner(knot, owner)) assert.NoError(t, e.AddRepo(owner, knot, repo)) assert.NoError(t, e.AddCollaborator(collaborator, knot, repo))
isOwner, err := e.IsKnotOwner(owner, knot) assert.NoError(t, err) assert.True(t, isOwner)
err = e.RemoveKnot(knot) assert.NoError(t, err)
isOwner, err = e.IsKnotOwner(owner, knot) assert.NoError(t, err) assert.False(t, isOwner)
assert.Empty(t, e.GetPermissionsInRepo(owner, knot, repo)) assert.Empty(t, e.GetPermissionsInRepo(collaborator, knot, repo))}
func TestRemoveSpindleOwner(t *testing.T) { e := setup(t)
err := e.AddSpindle("s.com") assert.NoError(t, err)
err = e.AddSpindleOwner("s.com", "did:plc:foo") assert.NoError(t, err)
spindles, err := e.GetSpindlesForUser("did:plc:foo") assert.NoError(t, err) assert.ElementsMatch(t, []string{ "s.com", }, spindles)
err = e.RemoveSpindleOwner("s.com", "did:plc:foo") assert.NoError(t, err)
spindles, err = e.GetSpindlesForUser("did:plc:foo") assert.NoError(t, err) assert.Empty(t, spindles)}
func TestRemoveSpindleMember(t *testing.T) { e := setup(t)
err := e.AddSpindle("s.com") assert.NoError(t, err)
err = e.AddSpindleOwner("s.com", "did:plc:foo") assert.NoError(t, err)
err = e.AddSpindleMember("s.com", "did:plc:bar") assert.NoError(t, err)
spindles, err := e.GetSpindlesForUser("did:plc:foo") assert.NoError(t, err) assert.ElementsMatch(t, []string{ "s.com", }, spindles)
spindles, err = e.GetSpindlesForUser("did:plc:bar") assert.NoError(t, err) assert.ElementsMatch(t, []string{ "s.com", }, spindles)
err = e.RemoveSpindleMember("s.com", "did:plc:bar") assert.NoError(t, err)
spindles, err = e.GetSpindlesForUser("did:plc:bar") assert.NoError(t, err) assert.Empty(t, spindles)}
func TestRemoveSpindle(t *testing.T) { e := setup(t)
err := e.AddSpindle("s.com") assert.NoError(t, err)
err = e.AddSpindleOwner("s.com", "did:plc:foo") assert.NoError(t, err)
err = e.AddSpindleMember("s.com", "did:plc:bar") assert.NoError(t, err)
users, err := e.GetSpindleUsersByRole("server:member", "s.com") assert.NoError(t, err) assert.ElementsMatch(t, []string{ "did:plc:foo", "did:plc:bar", }, users)
err = e.RemoveSpindle("s.com") assert.NoError(t, err)
// TODO: see this issue https://github.com/casbin/casbin/issues/1492 // s, err := e.E.GetAllDomains() // assert.Empty(t, s)
spindles, err := e.GetSpindleUsersByRole("server:member", "s.com") assert.NoError(t, err) assert.Empty(t, spindles)}
func TestWipeRepoPoliciesRemovesCollaborators(t *testing.T) { e := setup(t)
knot := "example.com" repo := "did:plc:akshay/my-repo" owner := "did:plc:akshay" collaborator := "did:plc:boltless"
_ = e.AddKnot(knot) _ = e.AddRepo(owner, knot, repo) err := e.AddCollaborator(collaborator, knot, repo) assert.NoError(t, err)
err = e.WipeRepoPolicies(knot, repo) assert.NoError(t, err)
assert.ElementsMatch(t, []string{}, e.GetPermissionsInRepo(collaborator, knot, repo), "collaborator policies must be wiped on repo teardown") assert.ElementsMatch(t, []string{}, e.GetPermissionsInRepo(owner, knot, repo), "owner policies must be wiped on repo teardown")}