astra/plc-mirror
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Implement commit signature validation

Browse source
This commit is contained in:
Max Ignatenko 2024-04-06 21:50:32 +01:00
parent f9dde4db39
commit ff0ea08296
9 changed files with 409 additions and 214 deletions
Download patch file Download diff file Expand all files Collapse all files

38
cmd/consumer/consumer.go
View file

@ -284,8 +284,21 @@ func (c *Consumer) processMessage(ctx context.Context, typ string, r io.Reader,
if err != nil {
return fmt.Errorf("repo.EnsureExists(%q): %w", payload.Repo, err)
}
if repoInfo.LastKnownKey == "" {
_, pubKey, err := resolver.GetPDSEndpointAndPublicKey(ctx, payload.Repo)
if err != nil {
return fmt.Errorf("failed to get DID doc for %q: %w", payload.Repo, err)
}
repoInfo.LastKnownKey = pubKey
err = c.db.Model(repoInfo).Where(&repo.Repo{ID: repoInfo.ID}).Updates(&repo.Repo{LastKnownKey: pubKey}).Error
if err != nil {
return fmt.Errorf("failed to update the key for %q: %w", payload.Repo, err)
}
}
if repoInfo.PDS != c.remote.ID {
u, err := resolver.GetPDSEndpoint(ctx, payload.Repo)
u, _, err := resolver.GetPDSEndpointAndPublicKey(ctx, payload.Repo)
if err == nil {
cur, err := pds.EnsureExists(ctx, c.db, u.String())
if err == nil {
@ -315,8 +328,6 @@ func (c *Consumer) processMessage(ctx context.Context, typ string, r io.Reader,
reposDiscovered.WithLabelValues(c.remote.Host).Inc()
}
// TODO: verify signature
expectRecords := false
deletions := []string{}
for _, op := range payload.Ops {
@ -345,10 +356,29 @@ func (c *Consumer) processMessage(ctx context.Context, typ string, r io.Reader,
}
}
newRecs, err := repo.ExtractRecords(ctx, bytes.NewReader(payload.Blocks))
newRecs, err := repo.ExtractRecords(ctx, bytes.NewReader(payload.Blocks), repoInfo.LastKnownKey)
if errors.Is(err, repo.ErrInvalidSignature) {
// Key might have been updated recently.
_, pubKey, err2 := resolver.GetPDSEndpointAndPublicKey(ctx, payload.Repo)
if err2 != nil {
return fmt.Errorf("failed to get DID doc for %q: %w", payload.Repo, err2)
}
if repoInfo.LastKnownKey != pubKey {
repoInfo.LastKnownKey = pubKey
err2 = c.db.Model(repoInfo).Where(&repo.Repo{ID: repoInfo.ID}).Updates(&repo.Repo{LastKnownKey: pubKey}).Error
if err2 != nil {
return fmt.Errorf("failed to update the key for %q: %w", payload.Repo, err2)
}
// Retry with the new key.
newRecs, err = repo.ExtractRecords(ctx, bytes.NewReader(payload.Blocks), pubKey)
}
}
if err != nil {
return fmt.Errorf("failed to extract records: %w", err)
}
recs := []repo.Record{}
for k, v := range newRecs {
parts := strings.SplitN(k, "/", 2)

2
cmd/consumer/relay_consumer.go
View file

@ -169,7 +169,7 @@ func (c *RelayConsumer) processMessage(ctx context.Context, typ string, r io.Rea
return nil
}
u, err := resolver.GetPDSEndpoint(ctx, did)
u, _, err := resolver.GetPDSEndpointAndPublicKey(ctx, did)
if err != nil {
return err
}

4
cmd/record-indexer/workerpool.go
View file

@ -137,7 +137,7 @@ func (p *WorkerPool) doWork(ctx context.Context, work WorkItem) error {
log := zerolog.Ctx(ctx)
defer close(work.signal)
u, err := resolver.GetPDSEndpoint(ctx, work.Repo.DID)
u, pubKey, err := resolver.GetPDSEndpointAndPublicKey(ctx, work.Repo.DID)
if err != nil {
return err
}
@ -217,7 +217,7 @@ retry:
return fmt.Errorf("failed to read 'rev' from the fetched repo: %w", err)
}
newRecs, err := repo.ExtractRecords(ctx, bytes.NewReader(b))
newRecs, err := repo.ExtractRecords(ctx, bytes.NewReader(b), pubKey)
if err != nil {
return fmt.Errorf("failed to extract records: %w", err)
}