mirror.go
5251 bytes
1package jobs
2
3import (
4 "context"
5 "fmt"
6 "path/filepath"
7 "runtime"
8 "strings"
9
10 "charm.land/log/v2"
11 "github.com/charmbracelet/soft-serve/git"
12 "github.com/charmbracelet/soft-serve/pkg/backend"
13 "github.com/charmbracelet/soft-serve/pkg/config"
14 "github.com/charmbracelet/soft-serve/pkg/db"
15 "github.com/charmbracelet/soft-serve/pkg/lfs"
16 "github.com/charmbracelet/soft-serve/pkg/ssrf"
17 "github.com/charmbracelet/soft-serve/pkg/store"
18 "github.com/charmbracelet/soft-serve/pkg/sync"
19)
20
21func init() {
22 Register("mirror-pull", mirrorPull{})
23}
24
25type mirrorPull struct{}
26
27// validateMirrorRemotes validates every remote configured on a mirror
28// repository and returns the git environment that must be applied to the sync
29// commands.
30//
31// `git remote update` fetches from all remotes, not just origin, so every one
32// of them has to pass. Any failure to read or validate skips the sync: a
33// remote that cannot be checked is not one to fetch from.
34func validateMirrorRemotes(r *git.Repository) ([]string, error) {
35 cfg, err := r.Config()
36 if err != nil {
37 return nil, fmt.Errorf("reading git config: %w", err)
38 }
39
40 var remotes []ssrf.ValidatedGitRemote
41 for _, sub := range cfg.Section("remote").Subsections {
42 url := sub.Option("url")
43 if url == "" {
44 continue
45 }
46
47 v, err := ssrf.ValidateGitRemote(url)
48 if err != nil {
49 return nil, fmt.Errorf("remote %q: %w", sub.Name, err)
50 }
51 remotes = append(remotes, v)
52 }
53
54 if len(remotes) == 0 {
55 return nil, fmt.Errorf("no remote url configured")
56 }
57
58 return ssrf.GitEnv(remotes...), nil
59}
60
61// Spec derives the spec used for pull mirrors and implements Runner.
62func (m mirrorPull) Spec(ctx context.Context) string {
63 cfg := config.FromContext(ctx)
64 if cfg.Jobs.MirrorPull != "" {
65 return cfg.Jobs.MirrorPull
66 }
67 return "@every 10m"
68}
69
70// Func runs the (pull) mirror job task and implements Runner.
71func (m mirrorPull) Func(ctx context.Context) func() {
72 cfg := config.FromContext(ctx)
73 logger := log.FromContext(ctx).WithPrefix("jobs.mirror")
74 b := backend.FromContext(ctx)
75 dbx := db.FromContext(ctx)
76 datastore := store.FromContext(ctx)
77 return func() {
78 repos, err := b.Repositories(ctx)
79 if err != nil {
80 logger.Error("error getting repositories", "err", err)
81 return
82 }
83
84 // Divide the work up among the number of CPUs.
85 wq := sync.NewWorkPool(ctx, runtime.GOMAXPROCS(0),
86 sync.WithWorkPoolLogger(logger.Errorf),
87 )
88
89 logger.Debug("updating mirror repos")
90 for _, repo := range repos {
91 if repo.IsMirror() {
92 r, err := repo.Open()
93 if err != nil {
94 logger.Error("error opening repository", "repo", repo.Name(), "err", err)
95 continue
96 }
97
98 name := repo.Name()
99 wq.Add(name, func() {
100 repo := repo
101
102 // Re-validate every configured remote before syncing.
103 // `remote update` touches all of them, and a remote may
104 // predate the import-time guard or have been written out
105 // of band. Validation failure skips the repo entirely.
106 remoteEnv, err := validateMirrorRemotes(r)
107 if err != nil {
108 logger.Warn("skipping mirror sync, remote failed validation", "repo", name, "err", err)
109 return
110 }
111
112 // remoteEnv carries the SSRF guard and must reach every
113 // command below, so build the full set once.
114 syncEnv := append(remoteEnv,
115 fmt.Sprintf(`GIT_SSH_COMMAND=ssh -o UserKnownHostsFile="%s" -o StrictHostKeyChecking=no -i "%s"`,
116 filepath.Join(cfg.DataPath, "ssh", "known_hosts"),
117 cfg.SSH.ClientKeyPath,
118 ),
119 )
120
121 cmds := []string{
122 "fetch --prune", // fetch prune before updating remote
123 "remote update --prune", // update remote and prune remote refs
124 }
125
126 for _, c := range cmds {
127 args := strings.Split(c, " ")
128 cmd := git.NewCommand(args...).WithContext(ctx).WithTimeout(-1)
129 cmd.AddEnvs(syncEnv...)
130
131 if _, err := cmd.RunInDir(r.Path); err != nil {
132 logger.Error("error running git remote update", "repo", name, "err", err)
133 }
134 }
135
136 if cfg.LFS.Enabled {
137 rcfg, err := r.Config()
138 if err != nil {
139 logger.Error("error getting git config", "repo", name, "err", err)
140 return
141 }
142
143 lfsEndpoint := rcfg.Section("lfs").Option("url")
144 if lfsEndpoint == "" {
145 // If there is no LFS url defined, means the repo
146 // doesn't use LFS and we can skip it.
147 return
148 }
149
150 // The endpoint is stored in the repo config and may
151 // predate validation, so check it before dialing.
152 if _, err := ssrf.ValidateGitRemote(lfsEndpoint); err != nil {
153 logger.Warn("skipping lfs sync, endpoint failed validation", "repo", name, "err", err)
154 return
155 }
156
157 ep, err := lfs.NewEndpoint(lfsEndpoint)
158 if err != nil {
159 logger.Error("error creating LFS endpoint", "repo", name, "err", err)
160 return
161 }
162
163 client := lfs.NewClient(ep)
164 if client == nil {
165 logger.Errorf("failed to create lfs client: unsupported endpoint %s", lfsEndpoint)
166 return
167 }
168
169 if err := backend.StoreRepoMissingLFSObjects(ctx, repo, dbx, datastore, client); err != nil {
170 logger.Error("failed to store missing lfs objects", "err", err, "path", r.Path)
171 return
172 }
173 }
174 })
175 }
176 }
177
178 wq.Run()
179 }
180}