1package replica
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "time"
8
9 "github.com/go-git/go-git/v5"
10 "github.com/go-git/go-git/v5/config"
11 "github.com/go-git/go-git/v5/plumbing"
12 "github.com/go-git/go-git/v5/plumbing/object"
13 "github.com/go-git/go-git/v5/plumbing/storer"
14 "github.com/go-git/go-git/v5/plumbing/transport"
15)
16
17type Remote struct {
18 URL string
19 Auth transport.AuthMethod
20}
21
22type Checkpoint struct {
23 Value string
24 Set bool
25}
26
27type Checkpoints interface {
28 ReadCheckpoint(ctx context.Context) (Checkpoint, error)
29 SwapCheckpoint(ctx context.Context, expected Checkpoint, next string) error
30}
31
32type Outcome int
33
34const (
35 Pushed Outcome = iota + 1
36 Adopted
37)
38
39type StepInput struct {
40 Repo *git.Repository
41 Dir string
42 Checkpoints Checkpoints
43 Primary Remote
44 Mirror *Remote
45 Signer *Signer
46 Blob []byte
47 When time.Time
48 Digest func(blob []byte) ([32]byte, error)
49 Current [32]byte
50}
51
52type Result struct {
53 Outcome Outcome
54 Commit plumbing.Hash
55 MirrorErr error
56 MirrorReseeded bool
57}
58
59var (
60 ErrRefused = errors.New("replica: refused")
61 ErrRemoteHasHistory = errors.New("replica: no checkpoint is set and the remote branch holds history")
62 ErrRemoteEmpty = errors.New("replica: a checkpoint is set and the remote branch is empty or absent")
63 ErrRewound = errors.New("replica: the remote branch is an ancestor of the checkpoint")
64 ErrForeignChild = errors.New("replica: the remote branch is a child of the checkpoint this appliance did not sign")
65 ErrNotLocalHead = errors.New("replica: the remote branch is a signed child of the checkpoint that is not the local branch head")
66 ErrDiverged = errors.New("replica: the remote branch is neither the checkpoint, an ancestor of it nor a child of it")
67 ErrOwnRootDiffers = errors.New("replica: no checkpoint is set and the remote branch is this appliance's own root, but its export differs from the store's")
68 ErrCheckpointMissing = errors.New("replica: the checkpoint commit is missing locally and is not on the remote branch")
69)
70
71var (
72 beforePush = func() {}
73 beforeMirror = func() {}
74 beforeReseed = func() {}
75)
76
77func refused(kind error, format string, args ...any) error {
78 return fmt.Errorf("%w: %w: "+format, append([]any{ErrRefused, kind}, args...)...)
79}
80
81func localHead(repo *git.Repository) (plumbing.Hash, error) {
82 ref, err := repo.Storer.Reference(Branch)
83 if errors.Is(err, plumbing.ErrReferenceNotFound) {
84 return plumbing.ZeroHash, nil
85 }
86 if err != nil {
87 return plumbing.ZeroHash, err
88 }
89 return ref.Hash(), nil
90}
91
92func hasCommit(repo *git.Repository, id plumbing.Hash) bool {
93 _, err := object.GetCommit(repo.Storer, id)
94 return err == nil
95}
96
97func Step(ctx context.Context, in StepInput) (Result, error) {
98 k, err := in.Checkpoints.ReadCheckpoint(ctx)
99 if err != nil {
100 return Result{}, err
101 }
102 if err := ensureCheckpoint(ctx, in.Repo, in.Dir, in.Primary, k); err != nil {
103 return Result{}, err
104 }
105 head, err := RemoteHead(ctx, in.Primary.URL, in.Primary.Auth, Branch)
106 if err != nil {
107 return Result{}, err
108 }
109 var parent plumbing.Hash
110 switch {
111 case !k.Set && head.IsZero():
112 case !k.Set:
113 return adoptRootOrRefuse(ctx, in, head)
114 case head.IsZero():
115 return Result{}, refused(ErrRemoteEmpty, "checkpoint %s", k.Value)
116 case head.String() == k.Value:
117 parent = head
118 default:
119 return adoptOrRefuse(ctx, in, k, head)
120 }
121 id, err := CommitFile(in.Repo, Branch, parent, in.Blob, in.When, in.Signer)
122 if err != nil {
123 return Result{}, err
124 }
125 beforePush()
126 if err := Push(ctx, in.Repo, in.Primary.URL, in.Primary.Auth, Branch, id, parent); err != nil {
127 return Result{}, err
128 }
129 if err := in.Checkpoints.SwapCheckpoint(ctx, k, id.String()); err != nil {
130 return Result{}, err
131 }
132 res := Result{Outcome: Pushed, Commit: id}
133 if in.Mirror != nil {
134 res.MirrorReseeded, res.MirrorErr = mirrorOrReseed(ctx, in, id)
135 }
136 return res, nil
137}
138
139func adoptOrRefuse(ctx context.Context, in StepInput, k Checkpoint, head plumbing.Hash) (Result, error) {
140 checkpoint := plumbing.NewHash(k.Value)
141 local, err := localHead(in.Repo)
142 if err != nil {
143 return Result{}, err
144 }
145 var objects storer.EncodedObjectStorer = in.Repo.Storer
146 if !hasCommit(in.Repo, head) {
147 s, done, err := scratch(in.Dir)
148 if err != nil {
149 return Result{}, err
150 }
151 defer done()
152 fetched, err := FetchHistory(ctx, s, in.Primary.URL, in.Primary.Auth, Branch)
153 if err != nil {
154 return Result{}, err
155 }
156 if fetched != head {
157 return Result{}, fmt.Errorf("%w: remote %s, read as %s", ErrRemoteMoved, fetched, head)
158 }
159 objects = s.Storer
160 }
161 c, err := object.GetCommit(objects, head)
162 if err != nil {
163 return Result{}, err
164 }
165 _, sigErr := VerifyCommit(objects, head, in.Signer.PublicKey())
166 if head == local && sigErr == nil && childOf(c, checkpoint) {
167 if err := in.Checkpoints.SwapCheckpoint(ctx, k, head.String()); err != nil {
168 return Result{}, err
169 }
170 return Result{Outcome: Adopted, Commit: head}, nil
171 }
172 return classify(objects, checkpoint, c, sigErr, local)
173}
174
175func childOf(c *object.Commit, parent plumbing.Hash) bool {
176 return len(c.ParentHashes) == 1 && c.ParentHashes[0] == parent
177}
178
179func classify(objects storer.EncodedObjectStorer, checkpoint plumbing.Hash, c *object.Commit, sigErr error, local plumbing.Hash) (Result, error) {
180 if kc, err := object.GetCommit(objects, checkpoint); err == nil {
181 if ok, err := c.IsAncestor(kc); err == nil && ok {
182 return Result{}, refused(ErrRewound, "remote %s, checkpoint %s", c.Hash, checkpoint)
183 }
184 }
185 switch {
186 case childOf(c, checkpoint) && sigErr != nil:
187 return Result{}, refused(ErrForeignChild, "remote %s: %w", c.Hash, signatureKind(sigErr))
188 case childOf(c, checkpoint):
189 return Result{}, refused(ErrNotLocalHead, "remote %s, local %s", c.Hash, local)
190 }
191 return Result{}, refused(ErrDiverged, "remote %s, checkpoint %s", c.Hash, checkpoint)
192}
193
194func signatureKind(err error) error {
195 for _, k := range []error{ErrUnsigned, ErrNotSSHSignature, ErrMalformed, ErrNamespace, ErrWrongKey, ErrBadSignature, ErrUnsupportedKey} {
196 if errors.Is(err, k) {
197 return k
198 }
199 }
200 return err
201}
202
203func adoptRootOrRefuse(ctx context.Context, in StepInput, head plumbing.Hash) (Result, error) {
204 local, err := localHead(in.Repo)
205 if err != nil {
206 return Result{}, err
207 }
208 if head != local {
209 return Result{}, refused(ErrRemoteHasHistory, "remote %s", head)
210 }
211 c, err := object.GetCommit(in.Repo.Storer, head)
212 if err != nil {
213 return Result{}, err
214 }
215 _, sigErr := VerifyCommit(in.Repo.Storer, head, in.Signer.PublicKey())
216 if c.NumParents() != 0 || sigErr != nil || in.Digest == nil {
217 return Result{}, refused(ErrRemoteHasHistory, "remote %s", head)
218 }
219 same, err := sameExport(in, head)
220 if err != nil {
221 return Result{}, refused(ErrOwnRootDiffers, "remote %s: %w", head, err)
222 }
223 if !same {
224 return Result{}, refused(ErrOwnRootDiffers, "remote %s", head)
225 }
226 if err := in.Checkpoints.SwapCheckpoint(ctx, Checkpoint{}, head.String()); err != nil {
227 return Result{}, err
228 }
229 return Result{Outcome: Adopted, Commit: head}, nil
230}
231
232func sameExport(in StepInput, id plumbing.Hash) (bool, error) {
233 blob, err := ReadFile(in.Repo.Storer, id)
234 if err != nil {
235 return false, err
236 }
237 committed, err := in.Digest(blob)
238 if err != nil {
239 return false, err
240 }
241 return committed == in.Current, nil
242}
243
244func mirrorOrReseed(ctx context.Context, in StepInput, id plumbing.Hash) (bool, error) {
245 beforeMirror()
246 err := pushMirror(ctx, in.Repo, *in.Mirror, id)
247 if !errors.Is(err, plumbing.ErrObjectNotFound) {
248 return false, err
249 }
250 return true, reseed(ctx, in, id)
251}
252
253func reseed(ctx context.Context, in StepInput, id plumbing.Hash) error {
254 beforeReseed()
255 s, done, err := scratch(in.Dir)
256 if err != nil {
257 return err
258 }
259 defer done()
260 head, err := FetchHistory(ctx, s, in.Primary.URL, in.Primary.Auth, Branch)
261 if err != nil {
262 return err
263 }
264 if head != id {
265 return fmt.Errorf("%w: re-seed: remote %s, checkpoint %s", ErrRemoteMoved, head, id)
266 }
267 return pushMirror(ctx, s, *in.Mirror, id)
268}
269
270func pushMirror(ctx context.Context, repo *git.Repository, mirror Remote, id plumbing.Hash) error {
271 head, err := RemoteHead(ctx, mirror.URL, mirror.Auth, Branch)
272 if err != nil {
273 return err
274 }
275 opts := &git.PushOptions{
276 RemoteName: remoteName,
277 RefSpecs: []config.RefSpec{config.RefSpec(id.String() + ":" + Branch.String())},
278 Auth: mirror.Auth,
279 ProxyOptions: proxyFor(mirror.Auth),
280 }
281 if !head.IsZero() {
282 opts.RequireRemoteRefs = []config.RefSpec{config.RefSpec(head.String() + ":" + Branch.String())}
283 }
284 return classifyPush(newRemote(repo, mirror.URL).PushContext(ctx, opts))
285}