diff --git a/appview/db/db.go b/appview/db/db.go --- a/appview/db/db.go +++ b/appview/db/db.go @@ -1961,6 +1961,22 @@ _, err := tx.Exec(`DROP TABLE IF EXISTS comments`) return err }) + orm.RunMigration(conn, logger, "add-knot-members-table", func(tx *sql.Tx) error { + _, err := tx.Exec(` + create table if not exists knot_members ( + id integer primary key autoincrement, + did text not null, + rkey text not null, + domain text not null, + subject text not null, + created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + unique (did, domain, subject) + ); + create index if not exists idx_knot_members_did_rkey on knot_members(did, rkey); + `) + return err + }) + return &DB{ db, logger, diff --git a/appview/db/registration.go b/appview/db/registration.go --- a/appview/db/registration.go +++ b/appview/db/registration.go @@ -92,6 +92,7 @@ func AddKnot(e Execer, domain, did string) error { _, err := e.Exec(` insert into registrations (domain, did) values (?, ?) + on conflict (domain, did) do nothing `, domain, did) return err } @@ -114,3 +115,90 @@ _, err := e.Exec(query, args...) return err } + +func AddKnotMember(e Execer, member models.KnotMember) error { + _, err := e.Exec( + `insert into knot_members (did, rkey, domain, subject) + values (?, ?, ?, ?) + on conflict (did, domain, subject) do update set rkey = excluded.rkey`, + member.Did, + member.Rkey, + member.Domain, + member.Subject, + ) + return err +} + +func RemoveKnotMember(e Execer, filters ...orm.Filter) error { + if len(filters) == 0 { + return fmt.Errorf("RemoveKnotMember requires at least one filter") + } + + var conditions []string + var args []any + for _, filter := range filters { + conditions = append(conditions, filter.Condition()) + args = append(args, filter.Arg()...) + } + + query := fmt.Sprintf(`delete from knot_members where %s`, strings.Join(conditions, " and ")) + + _, err := e.Exec(query, args...) + return err +} + +func GetKnotMembers(e Execer, filters ...orm.Filter) ([]models.KnotMember, error) { + var members []models.KnotMember + + var conditions []string + var args []any + for _, filter := range filters { + conditions = append(conditions, filter.Condition()) + args = append(args, filter.Arg()...) + } + + whereClause := "" + if conditions != nil { + whereClause = " where " + strings.Join(conditions, " and ") + } + + query := fmt.Sprintf( + `select id, did, rkey, domain, subject, created + from knot_members + %s + order by created + `, + whereClause, + ) + + rows, err := e.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + var member models.KnotMember + var createdAt string + + if err := rows.Scan( + &member.Id, + &member.Did, + &member.Rkey, + &member.Domain, + &member.Subject, + &createdAt, + ); err != nil { + return nil, err + } + + member.Created, err = time.Parse(time.RFC3339, createdAt) + if err != nil { + member.Created = time.Now() + } + + members = append(members, member) + } + + return members, nil +} diff --git a/appview/db/spindle.go b/appview/db/spindle.go --- a/appview/db/spindle.go +++ b/appview/db/spindle.go @@ -83,10 +83,10 @@ return spindles, nil } -// if there is an existing spindle with the same instance, this returns an error func AddSpindle(e Execer, spindle models.Spindle) error { _, err := e.Exec( - `insert into spindles (owner, instance) values (?, ?)`, + `insert into spindles (owner, instance) values (?, ?) + on conflict (owner, instance) do nothing`, spindle.Owner, spindle.Instance, ) @@ -147,6 +147,10 @@ return err } func RemoveSpindleMember(e Execer, filters ...orm.Filter) error { + if len(filters) == 0 { + return fmt.Errorf("RemoveSpindleMember requires at least one filter") + } + var conditions []string var args []any for _, filter := range filters { @@ -154,12 +158,7 @@ conditions = append(conditions, filter.Condition()) args = append(args, filter.Arg()...) } - whereClause := "" - if conditions != nil { - whereClause = " where " + strings.Join(conditions, " and ") - } - - query := fmt.Sprintf(`delete from spindle_members %s`, whereClause) + query := fmt.Sprintf(`delete from spindle_members where %s`, strings.Join(conditions, " and ")) _, err := e.Exec(query, args...) return err diff --git a/appview/ingester.go b/appview/ingester.go --- a/appview/ingester.go +++ b/appview/ingester.go @@ -38,6 +38,7 @@ "tangled.org/core/rbac" ) type Ingester struct { + Ctx context.Context Db *db.DB Enforcer *rbac.Enforcer IdResolver *idresolver.Resolver @@ -87,9 +88,9 @@ err = i.ingestSpindleMember(ctx, e) case tangled.SpindleNSID: err = i.ingestSpindle(ctx, e) case tangled.KnotMemberNSID: - err = i.ingestKnotMember(e) + err = i.ingestKnotMember(ctx, e) case tangled.KnotNSID: - err = i.ingestKnot(e) + err = i.ingestKnot(ctx, e) case tangled.StringNSID: err = i.ingestString(e) case tangled.RepoIssueNSID: @@ -640,7 +641,7 @@ l := i.Logger.With("handler", "ingestSpindleMember") l = l.With("nsid", e.Commit.Collection) switch e.Commit.Operation { - case jmodels.CommitOperationCreate: + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.SpindleMember{} err = json.Unmarshal(raw, &record) @@ -651,8 +652,20 @@ } // only spindle owner can invite to spindles ok, err := i.Enforcer.IsSpindleInviteAllowed(did, record.Instance) - if err != nil || !ok { - return fmt.Errorf("failed to enforce permissions: %w", err) + if err != nil { + return fmt.Errorf("failed to check invite permission: %w", err) + } + if !ok { + if verifyErr := i.verifySpindle(ctx, record.Instance, did); verifyErr != nil { + return fmt.Errorf("invite denied and verify failed: %w", verifyErr) + } + ok, err = i.Enforcer.IsSpindleInviteAllowed(did, record.Instance) + if err != nil { + return fmt.Errorf("failed to re-check invite permission: %w", err) + } + if !ok { + return fmt.Errorf("invite denied for did %s on spindle %s", did, record.Instance) + } } memberId, err := i.IdResolver.ResolveIdent(ctx, record.Subject) @@ -661,25 +674,71 @@ return err } if memberId.Handle.IsInvalidHandle() { - return err + return fmt.Errorf("invalid handle for member %s", record.Subject) } - err = db.AddSpindleMember(i.Db, models.SpindleMember{ + existing, err := db.GetSpindleMembers(i.Db, + orm.FilterEq("did", did), + orm.FilterEq("rkey", e.Commit.RKey), + ) + if err != nil { + return fmt.Errorf("failed to look up existing member: %w", err) + } + if len(existing) > 1 { + return fmt.Errorf("multiple spindle members with rkey %s", e.Commit.RKey) + } + + tx, err := i.Db.Begin() + if err != nil { + return fmt.Errorf("failed to start txn: %w", err) + } + committed := false + defer func() { + if committed { + return + } + tx.Rollback() + i.Enforcer.E.LoadPolicy() + }() + + if len(existing) == 1 { + prev := existing[0] + if prev.Instance != record.Instance || prev.Subject != memberId.DID { + if err = db.RemoveSpindleMember(tx, + orm.FilterEq("did", did), + orm.FilterEq("rkey", e.Commit.RKey), + ); err != nil { + return fmt.Errorf("failed to remove stale row: %w", err) + } + if err = i.Enforcer.RemoveSpindleMember(prev.Instance, prev.Subject.String()); err != nil { + return fmt.Errorf("failed to remove stale ACL: %w", err) + } + } + } + + if err = db.AddSpindleMember(tx, models.SpindleMember{ Did: syntax.DID(did), Rkey: e.Commit.RKey, Instance: record.Instance, Subject: memberId.DID, - }) - if !ok { + }); err != nil { return fmt.Errorf("failed to add to db: %w", err) } - err = i.Enforcer.AddSpindleMember(record.Instance, memberId.DID.String()) - if err != nil { + if err = i.Enforcer.AddSpindleMember(record.Instance, memberId.DID.String()); err != nil { return fmt.Errorf("failed to update ACLs: %w", err) } - l.Info("added spindle member") + if err = tx.Commit(); err != nil { + return fmt.Errorf("failed to commit txn: %w", err) + } + + if err = i.Enforcer.E.SavePolicy(); err != nil { + return fmt.Errorf("failed to save ACLs: %w", err) + } + committed = true + + l.Info("upserted spindle member") case jmodels.CommitOperationDelete: rkey := e.Commit.RKey @@ -698,6 +757,14 @@ tx, err := i.Db.Begin() if err != nil { return fmt.Errorf("failed to start txn: %w", err) } + committed := false + defer func() { + if committed { + return + } + tx.Rollback() + i.Enforcer.E.LoadPolicy() + }() // remove record by rkey && update enforcer if err = db.RemoveSpindleMember( @@ -721,6 +788,7 @@ if err = i.Enforcer.E.SavePolicy(); err != nil { return fmt.Errorf("failed to save ACLs: %w", err) } + committed = true l.Info("removed spindle member") } @@ -736,7 +804,7 @@ l := i.Logger.With("handler", "ingestSpindle") l = l.With("nsid", e.Commit.Collection) switch e.Commit.Operation { - case jmodels.CommitOperationCreate: + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.Spindle{} err = json.Unmarshal(raw, &record) @@ -756,19 +824,8 @@ l.Error("failed to add spindle to db", "err", err, "instance", instance) return err } - err = retry.Do( - func() error { return serververify.RunVerification(ctx, instance, did, i.Config.Core.Dev) }, - retry.Attempts(5), retry.Delay(5*time.Second), retry.MaxDelay(80*time.Second), - retry.DelayType(retry.BackOffDelay), retry.LastErrorOnly(true), - ) - if err != nil { - l.Error("failed to verify spindle after retries", "err", err, "instance", instance) - return err - } - - _, err = serververify.MarkSpindleVerified(i.Db, i.Enforcer, instance, did) - if err != nil { - return fmt.Errorf("failed to mark verified: %w", err) + if err := i.verifySpindle(ctx, instance, did); err != nil { + l.Warn("failed to verify spindle", "instance", instance, "did", did, "err", err) } return nil @@ -886,7 +943,7 @@ return nil } -func (i *Ingester) ingestKnotMember(e *jmodels.Event) error { +func (i *Ingester) ingestKnotMember(ctx context.Context, e *jmodels.Event) error { did := e.Did var err error @@ -894,7 +951,7 @@ l := i.Logger.With("handler", "ingestKnotMember") l = l.With("nsid", e.Commit.Collection) switch e.Commit.Operation { - case jmodels.CommitOperationCreate: + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.KnotMember{} err = json.Unmarshal(raw, &record) @@ -905,40 +962,154 @@ } // only knot owner can invite to knots ok, err := i.Enforcer.IsKnotInviteAllowed(did, record.Domain) - if err != nil || !ok { - return fmt.Errorf("failed to enforce permissions: %w", err) + if err != nil { + return fmt.Errorf("failed to check invite permission: %w", err) + } + if !ok { + if verifyErr := i.verifyKnot(ctx, record.Domain, did); verifyErr != nil { + return fmt.Errorf("invite denied and verify failed: %w", verifyErr) + } + ok, err = i.Enforcer.IsKnotInviteAllowed(did, record.Domain) + if err != nil { + return fmt.Errorf("failed to re-check invite permission: %w", err) + } + if !ok { + return fmt.Errorf("invite denied for did %s on knot %s", did, record.Domain) + } } - memberId, err := i.IdResolver.ResolveIdent(context.Background(), record.Subject) + memberId, err := i.IdResolver.ResolveIdent(ctx, record.Subject) if err != nil { return err } if memberId.Handle.IsInvalidHandle() { - return err + return fmt.Errorf("invalid handle for member %s", record.Subject) } - err = i.Enforcer.AddKnotMember(record.Domain, memberId.DID.String()) + existing, err := db.GetKnotMembers(i.Db, + orm.FilterEq("did", did), + orm.FilterEq("rkey", e.Commit.RKey), + ) if err != nil { + return fmt.Errorf("failed to look up existing member: %w", err) + } + if len(existing) > 1 { + return fmt.Errorf("multiple knot members with rkey %s", e.Commit.RKey) + } + + tx, err := i.Db.Begin() + if err != nil { + return fmt.Errorf("failed to start txn: %w", err) + } + committed := false + defer func() { + if committed { + return + } + tx.Rollback() + i.Enforcer.E.LoadPolicy() + }() + + if len(existing) == 1 { + prev := existing[0] + if prev.Domain != record.Domain || prev.Subject != memberId.DID { + if err = db.RemoveKnotMember(tx, + orm.FilterEq("did", did), + orm.FilterEq("rkey", e.Commit.RKey), + ); err != nil { + return fmt.Errorf("failed to remove stale row: %w", err) + } + if err = i.Enforcer.RemoveKnotMember(prev.Domain, prev.Subject.String()); err != nil { + return fmt.Errorf("failed to remove stale ACL: %w", err) + } + } + } + + if err = db.AddKnotMember(tx, models.KnotMember{ + Did: syntax.DID(did), + Rkey: e.Commit.RKey, + Domain: record.Domain, + Subject: memberId.DID, + }); err != nil { + return fmt.Errorf("failed to add to db: %w", err) + } + + if err = i.Enforcer.AddKnotMember(record.Domain, memberId.DID.String()); err != nil { return fmt.Errorf("failed to update ACLs: %w", err) } - l.Info("added knot member") + if err = tx.Commit(); err != nil { + return fmt.Errorf("failed to commit txn: %w", err) + } + + if err = i.Enforcer.E.SavePolicy(); err != nil { + return fmt.Errorf("failed to save ACLs: %w", err) + } + committed = true + + l.Info("upserted knot member") case jmodels.CommitOperationDelete: - // we don't store knot members in a table (like we do for spindle) - // and we can't remove this just yet. possibly fixed if we switch - // to either: - // 1. a knot_members table like with spindle and store the rkey - // 2. use the knot host as the rkey - // - // TODO: implement member deletion - l.Info("skipping knot member delete", "did", did, "rkey", e.Commit.RKey) + rkey := e.Commit.RKey + + members, err := db.GetKnotMembers( + i.Db, + orm.FilterEq("did", did), + orm.FilterEq("rkey", rkey), + ) + if err != nil { + return fmt.Errorf("failed to look up knot member with rkey %s: %w", rkey, err) + } + if len(members) == 0 { + l.Info("knot member already removed", "rkey", rkey) + return nil + } + if len(members) > 1 { + return fmt.Errorf("multiple knot members with rkey %s", rkey) + } + member := members[0] + + tx, err := i.Db.Begin() + if err != nil { + return fmt.Errorf("failed to start txn: %w", err) + } + committed := false + defer func() { + if committed { + return + } + tx.Rollback() + i.Enforcer.E.LoadPolicy() + }() + + if err = db.RemoveKnotMember( + tx, + orm.FilterEq("did", did), + orm.FilterEq("rkey", rkey), + ); err != nil { + return fmt.Errorf("failed to remove from db: %w", err) + } + + if err = i.Enforcer.RemoveKnotMember(member.Domain, member.Subject.String()); err != nil { + return fmt.Errorf("failed to update ACLs: %w", err) + } + + if err = tx.Commit(); err != nil { + return fmt.Errorf("failed to commit txn: %w", err) + } + + if err = i.Enforcer.E.SavePolicy(); err != nil { + return fmt.Errorf("failed to save ACLs: %w", err) + } + committed = true + + l.Info("removed knot member") } return nil } -func (i *Ingester) ingestKnot(e *jmodels.Event) error { +func (i *Ingester) ingestKnot(ctx context.Context, e *jmodels.Event) error { did := e.Did var err error @@ -946,7 +1117,7 @@ l := i.Logger.With("handler", "ingestKnot") l = l.With("nsid", e.Commit.Collection) switch e.Commit.Operation { - case jmodels.CommitOperationCreate: + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: raw := json.RawMessage(e.Commit.Record) record := tangled.Knot{} err = json.Unmarshal(raw, &record) @@ -963,21 +1134,8 @@ l.Error("failed to add knot to db", "err", err, "domain", domain) return err } - err = retry.Do( - func() error { - return serververify.RunVerification(context.Background(), domain, did, i.Config.Core.Dev) - }, - retry.Attempts(5), retry.Delay(5*time.Second), retry.MaxDelay(80*time.Second), - retry.DelayType(retry.BackOffDelay), retry.LastErrorOnly(true), - ) - if err != nil { - l.Error("failed to verify knot after retries", "err", err, "domain", domain) - return err - } - - err = serververify.MarkKnotVerified(i.Db, i.Enforcer, domain, did) - if err != nil { - return fmt.Errorf("failed to mark verified: %w", err) + if err := i.verifyKnot(ctx, domain, did); err != nil { + l.Warn("failed to verify knot", "domain", domain, "did", did, "err", err) } return nil @@ -1008,6 +1166,15 @@ tx.Rollback() i.Enforcer.E.LoadPolicy() }() + err = db.RemoveKnotMember( + tx, + orm.FilterEq("did", did), + orm.FilterEq("domain", domain), + ) + if err != nil { + return err + } + err = db.DeleteKnot( tx, orm.FilterEq("did", did), @@ -1037,6 +1204,113 @@ } return nil } + +const ( + verifyAttempts = 4 + verifyMinDelay = 1 * time.Second + verifyMaxDelay = 5 * time.Second +) + +func (i *Ingester) verifyKnot(ctx context.Context, domain, did string) error { + regs, err := db.GetRegistrations(i.Db, + orm.FilterEq("domain", domain), + orm.FilterEq("did", did), + ) + if err != nil { + return fmt.Errorf("look up registration: %w", err) + } + if len(regs) != 1 { + return fmt.Errorf("no registration for %s by %s", domain, did) + } + if regs[0].Registered != nil { + return nil + } + + err = retry.Do( + func() error { return serververify.RunVerification(ctx, domain, did, i.Config.Core.Dev) }, + retry.Context(ctx), + retry.Attempts(verifyAttempts), + retry.Delay(verifyMinDelay), + retry.MaxDelay(verifyMaxDelay), + retry.DelayType(retry.BackOffDelay), + retry.LastErrorOnly(true), + ) + if err != nil { + return fmt.Errorf("verify: %w", err) + } + return serververify.MarkKnotVerified(i.Db, i.Enforcer, domain, did) +} + +func (i *Ingester) verifySpindle(ctx context.Context, instance, did string) error { + spindles, err := db.GetSpindles(ctx, i.Db, + orm.FilterEq("instance", instance), + orm.FilterEq("owner", did), + ) + if err != nil { + return fmt.Errorf("look up spindle: %w", err) + } + if len(spindles) != 1 { + return fmt.Errorf("no spindle for %s by %s", instance, did) + } + if spindles[0].Verified != nil { + return nil + } + + err = retry.Do( + func() error { return serververify.RunVerification(ctx, instance, did, i.Config.Core.Dev) }, + retry.Context(ctx), + retry.Attempts(verifyAttempts), + retry.Delay(verifyMinDelay), + retry.MaxDelay(verifyMaxDelay), + retry.DelayType(retry.BackOffDelay), + retry.LastErrorOnly(true), + ) + if err != nil { + return fmt.Errorf("verify: %w", err) + } + _, err = serververify.MarkSpindleVerified(i.Db, i.Enforcer, instance, did) + return err +} + +const sweepConcurrency = 4 + +func (i *Ingester) SweepPendingVerifications() { + l := i.Logger.With("handler", "SweepPendingVerifications") + + var g errgroup.Group + g.SetLimit(sweepConcurrency) + + regs, err := db.GetRegistrations(i.Db, orm.FilterIs("registered", nil)) + if err != nil { + l.Error("failed to list unverified knots", "err", err) + } else { + for _, reg := range regs { + g.Go(func() error { + if err := i.verifyKnot(i.Ctx, reg.Domain, reg.ByDid); err != nil { + l.Warn("verify knot failed", "domain", reg.Domain, "did", reg.ByDid, "err", err) + } + return nil + }) + } + } + + spindles, err := db.GetSpindles(i.Ctx, i.Db, orm.FilterIs("verified", nil)) + if err != nil { + l.Error("failed to list unverified spindles", "err", err) + g.Wait() + return + } + for _, s := range spindles { + g.Go(func() error { + if err := i.verifySpindle(i.Ctx, s.Instance, s.Owner.String()); err != nil { + l.Warn("verify spindle failed", "instance", s.Instance, "owner", s.Owner, "err", err) + } + return nil + }) + } + g.Wait() +} + func (i *Ingester) ingestIssue(ctx context.Context, e *jmodels.Event) error { did := e.Did rkey := e.Commit.RKey diff --git a/appview/ingester_repo.go b/appview/ingester_repo.go --- a/appview/ingester_repo.go +++ b/appview/ingester_repo.go @@ -262,9 +262,39 @@ } return fmt.Errorf("failed to fetch repo for delete: %w", err) } - if err := db.RemoveRepo(i.Db, e.Did, e.Commit.RKey); err != nil { + if i.Enforcer == nil { + return fmt.Errorf("ingester has no RBAC enforcer configured") + } + + tx, err := i.Db.Begin() + if err != nil { + return fmt.Errorf("failed to start txn: %w", err) + } + committed := false + defer func() { + if committed { + return + } + tx.Rollback() + i.Enforcer.E.LoadPolicy() + }() + + if err := db.RemoveRepo(tx, e.Did, e.Commit.RKey); err != nil { return fmt.Errorf("failed to delete repo: %w", err) } + + if err := i.Enforcer.WipeRepoPolicies(repo.Knot, repo.RepoIdentifier()); err != nil { + return fmt.Errorf("failed to wipe repo permissions: %w", err) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("failed to commit txn: %w", err) + } + + if err := i.Enforcer.E.SavePolicy(); err != nil { + return fmt.Errorf("failed to save ACLs: %w", err) + } + committed = true i.Notifier.DeleteRepo(ctx, repo) l.Info("deleted repo row") diff --git a/appview/ingester_repo_test.go b/appview/ingester_repo_test.go --- a/appview/ingester_repo_test.go +++ b/appview/ingester_repo_test.go @@ -175,6 +175,19 @@ } } } +func assertNoRepoPolicies(t *testing.T, ing *Ingester, knot, repo string) { + t.Helper() + for _, perm := range []string{"repo:settings", "repo:push", "repo:owner", "repo:delete", "repo:invite", "repo:collaborator"} { + policies, err := ing.Enforcer.E.GetFilteredPolicy(1, knot, repo, perm) + if err != nil { + t.Fatalf("GetFilteredPolicy(%q): %v", perm, err) + } + if len(policies) != 0 { + t.Fatalf("expected no %s policies for %s, got %v", perm, repo, policies) + } + } +} + func TestIngestRepo_CreateInsertsNewRow(t *testing.T) { ing, spy := newTestIngester(t) @@ -418,6 +431,28 @@ ) if !errors.Is(err, sql.ErrNoRows) { t.Errorf("expected row to be deleted, got err = %v", err) } +} + +func TestIngestRepo_DeleteWipesRbac(t *testing.T) { + ing, _ := newTestIngester(t) + seedRepoRow(t, ing, "did:plc:akshay", "knot.example", "foo", "foo", "did:plc:repo1") + if err := ing.ensureRepoOwnerPermissions("did:plc:akshay", "knot.example", "did:plc:repo1"); err != nil { + t.Fatalf("ensureRepoOwnerPermissions: %v", err) + } + if err := ing.Enforcer.AddCollaborator("did:plc:boltless", "knot.example", "did:plc:repo1"); err != nil { + t.Fatalf("AddCollaborator: %v", err) + } + if err := ing.Enforcer.E.SavePolicy(); err != nil { + t.Fatalf("SavePolicy: %v", err) + } + assertRepoOwnerPermissions(t, ing, "did:plc:akshay", "knot.example", "did:plc:repo1") + + e := makeDeleteEvent("did:plc:akshay", "foo") + if err := ingestAcceptingOwner(t, ing, e); err != nil { + t.Fatalf("ingestRepo: %v", err) + } + + assertNoRepoPolicies(t, ing, "knot.example", "did:plc:repo1") } func TestIngestRepo_MalformedRecord(t *testing.T) { diff --git a/appview/knots/knots.go b/appview/knots/knots.go --- a/appview/knots/knots.go +++ b/appview/knots/knots.go @@ -1,6 +1,7 @@ package knots import ( + "context" "errors" "fmt" "log/slog" @@ -26,6 +27,7 @@ "tangled.org/core/rbac" "tangled.org/core/tid" comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/atclient" lexutil "github.com/bluesky-social/indigo/lex/util" ) @@ -165,26 +167,14 @@ l.Error("failed to start transaction", "err", err) fail() return } - defer func() { - tx.Rollback() - k.Enforcer.E.LoadPolicy() - }() + defer tx.Rollback() - err = db.AddKnot(tx, domain, user.Did) - if err != nil { + if err := db.AddKnot(tx, domain, user.Did); err != nil { l.Error("failed to insert", "err", err) fail() return } - err = k.Enforcer.AddKnot(domain) - if err != nil { - l.Error("failed to create knot", "err", err) - fail() - return - } - - // create record on pds client, err := k.OAuth.AuthorizedClient(r) if err != nil { l.Error("failed to authorize client", "err", err) @@ -198,7 +188,6 @@ if ex != nil { exCid = ex.Cid } - // re-announce by registering under same rkey _, err = comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.KnotNSID, Repo: user.Did, @@ -210,49 +199,20 @@ }, }, SwapRecord: exCid, }) - if err != nil { l.Error("failed to put record", "err", err) fail() return } - err = tx.Commit() - if err != nil { + if err := tx.Commit(); err != nil { l.Error("failed to commit transaction", "err", err) fail() return } - err = k.Enforcer.E.SavePolicy() - if err != nil { - l.Error("failed to update ACL", "err", err) - k.Pages.HxRefresh(w) - return - } + go k.Knotstream.AddSource(r.Context(), eventconsumer.NewKnotSource(domain)) - // begin verification - err = serververify.RunVerification(r.Context(), domain, user.Did, k.Config.Core.Dev) - if err != nil { - l.Error("verification failed", "err", err) - k.Pages.HxRefresh(w) - return - } - - err = serververify.MarkKnotVerified(k.Db, k.Enforcer, domain, user.Did) - if err != nil { - l.Error("failed to mark verified", "err", err) - k.Pages.HxRefresh(w) - return - } - - // add this knot to knotstream - go k.Knotstream.AddSource( - r.Context(), - eventconsumer.NewKnotSource(domain), - ) - - // ok k.Pages.HxRefresh(w) } @@ -553,16 +513,27 @@ k.Pages.Notice(w, noticeId, "Failed to add member, identity resolution failed.") return } - // write to pds client, err := k.OAuth.AuthorizedClient(r) if err != nil { l.Error("failed to authorize client", "err", err) fail() return } + + if err = k.Enforcer.AddKnotMember(domain, memberId.DID.String()); err != nil { + l.Error("failed to add member to ACLs", "err", err) + fail() + return + } + committed := false + defer func() { + if committed { + return + } + k.Enforcer.E.LoadPolicy() + }() rkey := tid.TID() - _, err = comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.KnotMemberNSID, Repo: user.Did, @@ -581,21 +552,13 @@ k.Pages.Notice(w, noticeId, "Failed to add record to PDS, try again later.") return } - err = k.Enforcer.AddKnotMember(domain, memberId.DID.String()) - if err != nil { - l.Error("failed to add member to ACLs", "err", err) - fail() - return - } - - err = k.Enforcer.E.SavePolicy() - if err != nil { + if err = k.Enforcer.E.SavePolicy(); err != nil { l.Error("failed to save ACL policy", "err", err) fail() return } + committed = true - // success k.Pages.HxRedirect(w, fmt.Sprintf("/settings/knots/%s", domain)) } @@ -649,33 +612,88 @@ k.Pages.Notice(w, noticeId, "Failed to remove member, identity resolution failed.") return } - // remove from enforcer - err = k.Enforcer.RemoveKnotMember(domain, memberId.DID.String()) + client, err := k.OAuth.AuthorizedClient(r) if err != nil { - l.Error("failed to update ACLs", "err", err) + l.Error("failed to authorize client", "err", err) fail() return } - client, err := k.OAuth.AuthorizedClient(r) + rkey, err := lookupKnotMemberRkey(r.Context(), k.Db, client, user.Did, domain, memberId.DID.String()) if err != nil { - l.Error("failed to authorize client", "err", err) + l.Warn("failed to look up member rkey", "err", err) + } + + if err = k.Enforcer.RemoveKnotMember(domain, memberId.DID.String()); err != nil { + l.Error("failed to update ACLs", "err", err) fail() return } + committed := false + defer func() { + if committed { + return + } + k.Enforcer.E.LoadPolicy() + }() - // TODO: We need to track the rkey for knot members to delete the record - // For now, just remove from ACLs - _ = client + if rkey != "" { + _, err = comatproto.RepoDeleteRecord(r.Context(), client, &comatproto.RepoDeleteRecord_Input{ + Collection: tangled.KnotMemberNSID, + Repo: user.Did, + Rkey: rkey, + }) + if err != nil { + l.Error("failed to delete record from PDS", "err", err) + k.Pages.Notice(w, noticeId, "Failed to delete record from PDS, try again later.") + return + } + } - // commit everything - err = k.Enforcer.E.SavePolicy() - if err != nil { + if err = k.Enforcer.E.SavePolicy(); err != nil { l.Error("failed to save ACLs", "err", err) fail() return } + committed = true - // ok k.Pages.HxRefresh(w) } + +func lookupKnotMemberRkey(ctx context.Context, d *db.DB, client *atclient.APIClient, ownerDid, domain, subject string) (string, error) { + members, err := db.GetKnotMembers( + d, + orm.FilterEq("did", ownerDid), + orm.FilterEq("domain", domain), + orm.FilterEq("subject", subject), + ) + if err != nil { + return "", fmt.Errorf("db lookup: %w", err) + } + if len(members) >= 1 { + return members[0].Rkey, nil + } + return findKnotMemberRkey(ctx, client, ownerDid, domain, subject, "") +} + +func findKnotMemberRkey(ctx context.Context, client *atclient.APIClient, repo, domain, subject, cursor string) (string, error) { + out, err := comatproto.RepoListRecords(ctx, client, tangled.KnotMemberNSID, cursor, 100, repo, false) + if err != nil { + return "", err + } + for _, rec := range out.Records { + m, ok := rec.Value.Val.(*tangled.KnotMember) + if !ok { + continue + } + if m.Domain != domain || m.Subject != subject { + continue + } + parts := strings.Split(rec.Uri, "/") + return parts[len(parts)-1], nil + } + if out.Cursor == nil || *out.Cursor == "" || *out.Cursor == cursor { + return "", nil + } + return findKnotMemberRkey(ctx, client, repo, domain, subject, *out.Cursor) +} diff --git a/appview/models/knot_member.go b/appview/models/knot_member.go new file mode 100644 --- /dev/null +++ b/appview/models/knot_member.go @@ -0,0 +1,16 @@ +package models + +import ( + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +type KnotMember struct { + Id int64 + Did syntax.DID + Rkey string + Domain string + Subject syntax.DID + Created time.Time +} diff --git a/appview/serververify/verify.go b/appview/serververify/verify.go --- a/appview/serververify/verify.go +++ b/appview/serververify/verify.go @@ -69,7 +69,11 @@ tx, err := d.Begin() if err != nil { return 0, fmt.Errorf("failed to create txn: %w", err) } + committed := false defer func() { + if committed { + return + } tx.Rollback() e.E.LoadPolicy() }() @@ -98,6 +102,7 @@ err = e.E.SavePolicy() if err != nil { return 0, fmt.Errorf("failed to update ACL: %w", err) } + committed = true return rowId, nil } @@ -108,7 +113,11 @@ tx, err := d.BeginTx(context.Background(), nil) if err != nil { return fmt.Errorf("failed to start tx: %w", err) } + committed := false defer func() { + if committed { + return + } tx.Rollback() e.E.LoadPolicy() }() @@ -144,6 +153,7 @@ err = e.E.SavePolicy() if err != nil { return fmt.Errorf("failed to update ACLs: %w", err) } + committed = true return nil } diff --git a/appview/spindles/spindles.go b/appview/spindles/spindles.go --- a/appview/spindles/spindles.go +++ b/appview/spindles/spindles.go @@ -170,29 +170,17 @@ l.Error("failed to start transaction", "err", err) fail() return } - defer func() { - tx.Rollback() - s.Enforcer.E.LoadPolicy() - }() + defer tx.Rollback() - err = db.AddSpindle(tx, models.Spindle{ + if err := db.AddSpindle(tx, models.Spindle{ Owner: syntax.DID(user.Did), Instance: instance, - }) - if err != nil { + }); err != nil { l.Error("failed to insert", "err", err) fail() return } - err = s.Enforcer.AddSpindle(instance) - if err != nil { - l.Error("failed to create spindle", "err", err) - fail() - return - } - - // create record on pds client, err := s.OAuth.AuthorizedClient(r) if err != nil { l.Error("failed to authorize client", "err", err) @@ -206,7 +194,6 @@ if ex != nil { exCid = ex.Cid } - // re-announce by registering under same rkey _, err = comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.SpindleNSID, Repo: user.Did, @@ -218,43 +205,18 @@ }, }, SwapRecord: exCid, }) - if err != nil { l.Error("failed to put record", "err", err) fail() return } - err = tx.Commit() - if err != nil { + if err := tx.Commit(); err != nil { l.Error("failed to commit transaction", "err", err) fail() return } - err = s.Enforcer.E.SavePolicy() - if err != nil { - l.Error("failed to update ACL", "err", err) - s.Pages.HxRefresh(w) - return - } - - // begin verification - err = serververify.RunVerification(r.Context(), instance, user.Did, s.Config.Core.Dev) - if err != nil { - l.Error("verification failed", "err", err) - s.Pages.HxRefresh(w) - return - } - - _, err = serververify.MarkSpindleVerified(s.Db, s.Enforcer, instance, user.Did) - if err != nil { - l.Error("failed to mark verified", "err", err) - s.Pages.HxRefresh(w) - return - } - - // ok s.Pages.HxRefresh(w) } diff --git a/appview/state/state.go b/appview/state/state.go --- a/appview/state/state.go +++ b/appview/state/state.go @@ -174,6 +174,7 @@ notifier := notify.NewMergedNotifier(notifiers) notifier = lognotify.NewLoggingNotifier(notifier, tlog.SubLogger(logger, "notify")) ingester := appview.Ingester{ + Ctx: ctx, Db: d, Enforcer: enforcer, IdResolver: res, @@ -188,6 +189,8 @@ err = jc.StartJetstream(ctx, ingester.Ingest()) if err != nil { return nil, fmt.Errorf("failed to start jetstream watcher: %w", err) } + + go ingester.SweepPendingVerifications() var cfClient *cloudflare.Client if config.Cloudflare.ApiToken != "" {