patch of knotserver,spindle: ingest collaborator record and update rbac · round #0 · pull #398 · tangled.org/core

+13

jetstream/jetstream.go

···

       52
       52
        
       	j.mu.Unlock()

     

       53
       53
        
       }

     

       54
       54
        
       

     

       55
       55
       +
       func (j *JetstreamClient) RemoveDid(did string) {

     

       56
       56
       +
       	if did == "" {

     

       57
       57
       +
       		return

     

       58
       58
       +
       	}

     

       59
       59
       +
       

     

       60
       60
       +
       	if j.logDids {

     

       61
       61
       +
       		j.l.Info("removing did from in-memory filter", "did", did)

     

       62
       62
       +
       	}

     

       63
       63
       +
       	j.mu.Lock()

     

       64
       64
       +
       	delete(j.wantedDids, did)

     

       65
       65
       +
       	j.mu.Unlock()

     

       66
       66
       +
       }

     

       67
       67
       +
       

     

       55
       68
        
       type processor func(context.Context, *models.Event) error

     

       56
       69
        
       

     

       57
       70
        
       func (j *JetstreamClient) withDidFilter(processFunc processor) processor {

+61

knotserver/ingester.go

···

       213
       213
        
       	return h.db.InsertEvent(event, h.n)

     

       214
       214
        
       }

     

       215
       215
        
       

     

       216
       216
       +
       // duplicated from add collaborator

     

       217
       217
       +
       func (h *Handle) processCollaborator(ctx context.Context, did string, record tangled.RepoCollaborator) error {

     

       218
       218
       +
       	repoAt, err := syntax.ParseATURI(record.Repo)

     

       219
       219
       +
       	if err != nil {

     

       220
       220
       +
       		return err

     

       221
       221
       +
       	}

     

       222
       222
       +
       

     

       223
       223
       +
       	resolver := idresolver.DefaultResolver()

     

       224
       224
       +
       

     

       225
       225
       +
       	subjectId, err := resolver.ResolveIdent(ctx, record.Subject)

     

       226
       226
       +
       	if err != nil || subjectId.Handle.IsInvalidHandle() {

     

       227
       227
       +
       		return err

     

       228
       228
       +
       	}

     

       229
       229
       +
       

     

       230
       230
       +
       	// TODO: fix this for good, we need to fetch the record here unfortunately

     

       231
       231
       +
       	// resolve this aturi to extract the repo record

     

       232
       232
       +
       	owner, err := resolver.ResolveIdent(ctx, repoAt.Authority().String())

     

       233
       233
       +
       	if err != nil || owner.Handle.IsInvalidHandle() {

     

       234
       234
       +
       		return fmt.Errorf("failed to resolve handle: %w", err)

     

       235
       235
       +
       	}

     

       236
       236
       +
       

     

       237
       237
       +
       	xrpcc := xrpc.Client{

     

       238
       238
       +
       		Host: owner.PDSEndpoint(),

     

       239
       239
       +
       	}

     

       240
       240
       +
       

     

       241
       241
       +
       	resp, err := comatproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String())

     

       242
       242
       +
       	if err != nil {

     

       243
       243
       +
       		return err

     

       244
       244
       +
       	}

     

       245
       245
       +
       

     

       246
       246
       +
       	repo := resp.Value.Val.(*tangled.Repo)

     

       247
       247
       +
       	didSlashRepo, _ := securejoin.SecureJoin(owner.DID.String(), repo.Name)

     

       248
       248
       +
       

     

       249
       249
       +
       	// check perms for this user

     

       250
       250
       +
       	if ok, err := h.e.IsCollaboratorInviteAllowed(owner.DID.String(), rbac.ThisServer, didSlashRepo); !ok || err != nil {

     

       251
       251
       +
       		return fmt.Errorf("insufficient permissions: %w", err)

     

       252
       252
       +
       	}

     

       253
       253
       +
       

     

       254
       254
       +
       	if err := h.db.AddDid(subjectId.DID.String()); err != nil {

     

       255
       255
       +
       		return err

     

       256
       256
       +
       	}

     

       257
       257
       +
       	h.jc.AddDid(subjectId.DID.String())

     

       258
       258
       +
       

     

       259
       259
       +
       	if err := h.e.AddCollaborator(subjectId.DID.String(), rbac.ThisServer, didSlashRepo); err != nil {

     

       260
       260
       +
       		return err

     

       261
       261
       +
       	}

     

       262
       262
       +
       

     

       263
       263
       +
       	return h.fetchAndAddKeys(ctx, subjectId.DID.String())

     

       264
       264
       +
       }

     

       265
       265
       +
       

     

       216
       266
        
       func (h *Handle) fetchAndAddKeys(ctx context.Context, did string) error {

     

       217
       267
        
       	l := log.FromContext(ctx)

     

       218
       268
        
       

     
···

       292
       342
        
       		if err := h.processKnotMember(ctx, did, record); err != nil {

     

       293
       343
        
       			return fmt.Errorf("failed to process knot member: %w", err)

     

       294
       344
        
       		}

     

       345
       345
       +
       

     

       295
       346
        
       	case tangled.RepoPullNSID:

     

       296
       347
        
       		var record tangled.RepoPull

     

       297
       348
        
       		if err := json.Unmarshal(raw, &record); err != nil {

     
···

       300
       351
        
       		if err := h.processPull(ctx, did, record); err != nil {

     

       301
       352
        
       			return fmt.Errorf("failed to process knot member: %w", err)

     

       302
       353
        
       		}

     

       354
       354
       +
       

     

       355
       355
       +
       	case tangled.RepoCollaboratorNSID:

     

       356
       356
       +
       		var record tangled.RepoCollaborator

     

       357
       357
       +
       		if err := json.Unmarshal(raw, &record); err != nil {

     

       358
       358
       +
       			return fmt.Errorf("failed to unmarshal record: %w", err)

     

       359
       359
       +
       		}

     

       360
       360
       +
       		if err := h.processCollaborator(ctx, did, record); err != nil {

     

       361
       361
       +
       			return fmt.Errorf("failed to process knot member: %w", err)

     

       362
       362
       +
       		}

     

       363
       363
       +
       

     

       303
       364
        
       	}

     

       304
       365
        
       

     

       305
       366
        
       	return err

knotserver/server.go

···

       76
       76
        
       		tangled.PublicKeyNSID,

     

       77
       77
        
       		tangled.KnotMemberNSID,

     

       78
       78
        
       		tangled.RepoPullNSID,

     

       79
       79
       +
       		tangled.RepoCollaboratorNSID,

     

       79
       80
        
       	}, nil, logger, db, true, c.Server.LogDids)

     

       80
       81
        
       	if err != nil {

     

       81
       82
        
       		logger.Error("failed to setup jetstream", "error", err)

+103

spindle/ingester.go

···

       8
       8
        
       

     

       9
       9
        
       	"tangled.sh/tangled.sh/core/api/tangled"

     

       10
       10
        
       	"tangled.sh/tangled.sh/core/eventconsumer"

     

       11
       11
       +
       	"tangled.sh/tangled.sh/core/idresolver"

     

       11
       12
        
       	"tangled.sh/tangled.sh/core/rbac"

     

       12
       13
        
       

     

       14
       14
       +
       	comatproto "github.com/bluesky-social/indigo/api/atproto"

     

       15
       15
       +
       	"github.com/bluesky-social/indigo/atproto/identity"

     

       16
       16
       +
       	"github.com/bluesky-social/indigo/atproto/syntax"

     

       17
       17
       +
       	"github.com/bluesky-social/indigo/xrpc"

     

       13
       18
        
       	"github.com/bluesky-social/jetstream/pkg/models"

     

       19
       19
       +
       	securejoin "github.com/cyphar/filepath-securejoin"

     

       14
       20
        
       )

     

       15
       21
        
       

     

       16
       22
        
       type Ingester func(ctx context.Context, e *models.Event) error

     
···

       35
       41
        
       			s.ingestMember(ctx, e)

     

       36
       42
        
       		case tangled.RepoNSID:

     

       37
       43
        
       			s.ingestRepo(ctx, e)

     

       44
       44
       +
       		case tangled.RepoCollaboratorNSID:

     

       45
       45
       +
       			s.ingestCollaborator(ctx, e)

     

       38
       46
        
       		}

     

       39
       47
        
       

     

       40
       48
        
       		return err

     
···

       144
       152
        
       	}

     

       145
       153
        
       	return nil

     

       146
       154
        
       }

     

       155
       155
       +
       

     

       156
       156
       +
       func (s *Spindle) ingestCollaborator(ctx context.Context, e *models.Event) error {

     

       157
       157
       +
       	var err error

     

       158
       158
       +
       

     

       159
       159
       +
       	l := s.l.With("component", "ingester", "record", tangled.RepoCollaboratorNSID, "did", e.Did)

     

       160
       160
       +
       

     

       161
       161
       +
       	l.Info("ingesting collaborator record")

     

       162
       162
       +
       

     

       163
       163
       +
       	switch e.Commit.Operation {

     

       164
       164
       +
       	case models.CommitOperationCreate, models.CommitOperationUpdate:

     

       165
       165
       +
       		raw := e.Commit.Record

     

       166
       166
       +
       		record := tangled.RepoCollaborator{}

     

       167
       167
       +
       		err = json.Unmarshal(raw, &record)

     

       168
       168
       +
       		if err != nil {

     

       169
       169
       +
       			l.Error("invalid record", "error", err)

     

       170
       170
       +
       			return err

     

       171
       171
       +
       		}

     

       172
       172
       +
       

     

       173
       173
       +
       		resolver := idresolver.DefaultResolver()

     

       174
       174
       +
       

     

       175
       175
       +
       		subjectId, err := resolver.ResolveIdent(ctx, record.Subject)

     

       176
       176
       +
       		if err != nil || subjectId.Handle.IsInvalidHandle() {

     

       177
       177
       +
       			return err

     

       178
       178
       +
       		}

     

       179
       179
       +
       

     

       180
       180
       +
       		repoAt, err := syntax.ParseATURI(record.Repo)

     

       181
       181
       +
       		if err != nil {

     

       182
       182
       +
       			l.Info("rejecting record, invalid repoAt", "repoAt", record.Repo)

     

       183
       183
       +
       			return nil

     

       184
       184
       +
       		}

     

       185
       185
       +
       

     

       186
       186
       +
       		// TODO: get rid of this entirely

     

       187
       187
       +
       		// resolve this aturi to extract the repo record

     

       188
       188
       +
       		owner, err := resolver.ResolveIdent(ctx, repoAt.Authority().String())

     

       189
       189
       +
       		if err != nil || owner.Handle.IsInvalidHandle() {

     

       190
       190
       +
       			return fmt.Errorf("failed to resolve handle: %w", err)

     

       191
       191
       +
       		}

     

       192
       192
       +
       

     

       193
       193
       +
       		xrpcc := xrpc.Client{

     

       194
       194
       +
       			Host: owner.PDSEndpoint(),

     

       195
       195
       +
       		}

     

       196
       196
       +
       

     

       197
       197
       +
       		resp, err := comatproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String())

     

       198
       198
       +
       		if err != nil {

     

       199
       199
       +
       			return err

     

       200
       200
       +
       		}

     

       201
       201
       +
       

     

       202
       202
       +
       		repo := resp.Value.Val.(*tangled.Repo)

     

       203
       203
       +
       		didSlashRepo, _ := securejoin.SecureJoin(owner.DID.String(), repo.Name)

     

       204
       204
       +
       

     

       205
       205
       +
       		// check perms for this user

     

       206
       206
       +
       		if ok, err := s.e.IsCollaboratorInviteAllowed(owner.DID.String(), rbac.ThisServer, didSlashRepo); !ok || err != nil {

     

       207
       207
       +
       			return fmt.Errorf("insufficient permissions: %w", err)

     

       208
       208
       +
       		}

     

       209
       209
       +
       

     

       210
       210
       +
       		// add collaborator to rbac

     

       211
       211
       +
       		if err := s.e.AddCollaborator(record.Subject, rbac.ThisServer, didSlashRepo); err != nil {

     

       212
       212
       +
       			l.Error("failed to add repo to enforcer", "error", err)

     

       213
       213
       +
       			return fmt.Errorf("failed to add repo: %w", err)

     

       214
       214
       +
       		}

     

       215
       215
       +
       

     

       216
       216
       +
       		return nil

     

       217
       217
       +
       	}

     

       218
       218
       +
       	return nil

     

       219
       219
       +
       }

     

       220
       220
       +
       

     

       221
       221
       +
       func (s *Spindle) fetchAndAddCollaborators(ctx context.Context, owner *identity.Identity, didSlashRepo string) error {

     

       222
       222
       +
       	l := s.l.With("component", "ingester", "handler", "fetchAndAddCollaborators")

     

       223
       223
       +
       

     

       224
       224
       +
       	l.Info("fetching and adding existing collaborators")

     

       225
       225
       +
       

     

       226
       226
       +
       	xrpcc := xrpc.Client{

     

       227
       227
       +
       		Host: owner.PDSEndpoint(),

     

       228
       228
       +
       	}

     

       229
       229
       +
       

     

       230
       230
       +
       	resp, err := comatproto.RepoListRecords(ctx, &xrpcc, tangled.RepoCollaboratorNSID, "", 50, owner.DID.String(), false)

     

       231
       231
       +
       	if err != nil {

     

       232
       232
       +
       		return err

     

       233
       233
       +
       	}

     

       234
       234
       +
       

     

       235
       235
       +
       	var errs error

     

       236
       236
       +
       	for _, r := range resp.Records {

     

       237
       237
       +
       		if r == nil {

     

       238
       238
       +
       			continue

     

       239
       239
       +
       		}

     

       240
       240
       +
       		record := r.Value.Val.(*tangled.RepoCollaborator)

     

       241
       241
       +
       

     

       242
       242
       +
       		if err := s.e.AddCollaborator(record.Subject, rbac.ThisServer, didSlashRepo); err != nil {

     

       243
       243
       +
       			l.Error("failed to add repo to enforcer", "error", err)

     

       244
       244
       +
       			errors.Join(errs, fmt.Errorf("failed to add repo: %w", err))

     

       245
       245
       +
       		}

     

       246
       246
       +
       	}

     

       247
       247
       +
       

     

       248
       248
       +
       	return errs

     

       249
       249
       +
       }

spindle/server.go

···

       111
       111
        
       	collections := []string{

     

       112
       112
        
       		tangled.SpindleMemberNSID,

     

       113
       113
        
       		tangled.RepoNSID,

     

       114
       114
       +
       		tangled.RepoCollaboratorNSID,

     

       114
       115
        
       	}

     

       115
       116
        
       	jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, logger, d, true, true)

     

       116
       117
        
       	if err != nil {