internal/mirror/mirror.go
215 lines · 6297 bytes
1// Package mirror synchronizes repositories with foreign remotes: push
2// mirrors propagate local refs outward after each receive, pull mirrors
3// keep a local copy fresh from an upstream. Sync runs in a background
4// worker, never in the push path; outcomes are recorded per mirror so
5// `repo mirror list` shows failure states like webhook deliveries do.
6package mirror
7
8import (
9 "context"
10 "fmt"
11 "log/slog"
12 "net"
13 "net/url"
14 "os"
15 "os/exec"
16 "path/filepath"
17 "strconv"
18 "strings"
19 "time"
20
21 "gitbay.org/gitbay/internal/config"
22 "gitbay.org/gitbay/internal/control"
23 "gitbay.org/gitbay/internal/store"
24 "gitbay.org/gitbay/internal/toolpath"
25 "gitbay.org/gitbay/internal/webhook"
26)
27
28const askpassScript = `#!/bin/sh
29case "$1" in
30 Username*) echo "${GITBAY_MIRROR_USER}" ;;
31 *) echo "${GITBAY_MIRROR_TOKEN}" ;;
32esac
33`
34
35type Worker struct {
36 St *store.Store
37 Cfg config.Config
38 Tick time.Duration
39 // Lookup resolves a mirror's host immediately before each sync.
40 Lookup func(ctx context.Context, host string) ([]net.IP, error)
41 // gitErr is set when the server's git cannot pin addresses; no
42 // mirror syncs while it is.
43 gitErr error
44}
45
46func New(st *store.Store, cfg config.Config) *Worker {
47 tick := 10 * time.Second
48 if v := os.Getenv("GITBAY_MIRROR_TICK"); v != "" {
49 if d, err := time.ParseDuration(v); err == nil {
50 tick = d
51 }
52 }
53 return &Worker{St: st, Cfg: cfg, Tick: tick,
54 Lookup: func(ctx context.Context, host string) ([]net.IP, error) {
55 return net.DefaultResolver.LookupIP(ctx, "ip", host)
56 }}
57}
58
59func (w *Worker) Run(ctx context.Context) {
60 out, err := exec.CommandContext(ctx, toolpath.Look("git"), "version").Output()
61 if err != nil {
62 w.gitErr = fmt.Errorf("mirrors disabled: running git version: %v", err)
63 } else {
64 w.gitErr = gitVersionOK(string(out))
65 }
66 if w.gitErr != nil {
67 slog.Error("mirror: not syncing", "err", w.gitErr)
68 }
69 t := time.NewTicker(w.Tick)
70 defer t.Stop()
71 for {
72 select {
73 case <-ctx.Done():
74 return
75 case <-t.C:
76 w.sweep()
77 }
78 }
79}
80
81func (w *Worker) sweep() {
82 interval := w.Cfg.Mirrors.PullIntervalMinutes * 60
83 due, err := w.St.DueMirrors(interval)
84 if err != nil {
85 slog.Error("mirror: listing due", "err", err)
86 return
87 }
88 for _, m := range due {
89 if w.gitErr != nil {
90 w.St.SetMirrorResult(m.ID, w.gitErr.Error())
91 continue
92 }
93 if err := w.sync(m); err != nil {
94 slog.Warn("mirror sync failed", "mirror", m.ID, "url", m.URL, "err", err)
95 w.St.SetMirrorResult(m.ID, err.Error())
96 } else {
97 w.St.SetMirrorResult(m.ID, "")
98 }
99 }
100}
101
102func (w *Worker) sync(m store.Mirror) error {
103 repo, err := w.St.RepoByID(m.RepoID)
104 if err != nil {
105 return err
106 }
107 dir := control.RepoDir(w.Cfg.Server.Root, repo.OwnerName, repo.Name)
108 u, err := url.Parse(m.URL)
109 if err != nil {
110 return err
111 }
112 if u.Scheme != "https" && u.Scheme != "http" {
113 return fmt.Errorf("mirror URL scheme %q is not http or https", u.Scheme)
114 }
115
116 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
117 defer cancel()
118 // The URL was checked when saved, but DNS can answer differently
119 // now. Check what it resolves to at sync time, then let git connect
120 // to exactly those addresses.
121 ips, err := w.Lookup(ctx, u.Hostname())
122 if err != nil {
123 return fmt.Errorf("resolving %s: %w", u.Hostname(), err)
124 }
125 if len(ips) == 0 {
126 // An empty resolve list would leave curl to resolve the host itself.
127 return fmt.Errorf("%s resolves to no address", u.Hostname())
128 }
129 if err := webhook.CheckAddrs(u.Hostname(), ips, w.Cfg.Webhooks.AllowLocal); err != nil {
130 return err
131 }
132
133 // No system or global gitconfig: a proxy, URL rewrite or redirect
134 // setting there would take git around the pin.
135 env := []string{"GIT_TERMINAL_PROMPT=0", "HOME=" + w.Cfg.Server.Root,
136 "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL=/dev/null"}
137 if m.Token != "" {
138 askpass := filepath.Join(w.Cfg.Server.Root, "mirror-askpass.sh")
139 if err := os.WriteFile(askpass, []byte(askpassScript), 0o700); err != nil {
140 return err
141 }
142 user := m.Username
143 if user == "" {
144 user = "x-access-token"
145 }
146 env = append(env,
147 "GIT_ASKPASS="+askpass,
148 "GITBAY_MIRROR_USER="+user,
149 "GITBAY_MIRROR_TOKEN="+m.Token)
150 }
151
152 args := append(pinArgs(u, ips), "-C", dir)
153 if m.Direction == "push" {
154 // Branches and tags only: internal refs (merge-requests) stay home.
155 args = append(args, "push", "--prune", m.URL,
156 "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*")
157 } else {
158 args = append(args, "fetch", "--prune", m.URL,
159 "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*")
160 }
161 cmd := exec.CommandContext(ctx, toolpath.Look("git"), args...)
162 cmd.Env = env
163 if out, err := cmd.CombinedOutput(); err != nil {
164 return fmt.Errorf("git %s: %v: %.300s", m.Direction, err, out)
165 }
166 return nil
167}
168
169// pinArgs keeps git on the addresses just checked: curl's resolve list
170// pins the host, and with redirects off a server cannot send git on to
171// a host nobody checked. An address literal needs no pin.
172func pinArgs(u *url.URL, ips []net.IP) []string {
173 args := []string{"-c", "http.followRedirects=false"}
174 host := u.Hostname()
175 if net.ParseIP(host) != nil {
176 return args
177 }
178 port := u.Port()
179 if port == "" {
180 port = "443"
181 if u.Scheme == "http" {
182 port = "80"
183 }
184 }
185 addrs := make([]string, len(ips))
186 for i, ip := range ips {
187 if ip.To4() == nil {
188 addrs[i] = "[" + ip.String() + "]"
189 } else {
190 addrs[i] = ip.String()
191 }
192 }
193 return append(args, "-c", "http.curloptResolve="+host+":"+port+":"+strings.Join(addrs, ","))
194}
195
196// gitVersionOK accepts the output of `git version` for git 2.37 or
197// later, the first release with http.curloptResolve. An older git
198// ignores the setting and would resolve the host itself.
199func gitVersionOK(out string) error {
200 fields := strings.Fields(out)
201 if len(fields) >= 3 && fields[0] == "git" && fields[1] == "version" {
202 parts := strings.Split(fields[2], ".")
203 if len(parts) >= 2 {
204 major, err1 := strconv.Atoi(parts[0])
205 minor, err2 := strconv.Atoi(parts[1])
206 if err1 == nil && err2 == nil {
207 if major > 2 || major == 2 && minor >= 37 {
208 return nil
209 }
210 return fmt.Errorf("mirrors disabled: git %s is older than 2.37 and cannot pin mirror addresses", fields[2])
211 }
212 }
213 }
214 return fmt.Errorf("mirrors disabled: cannot read git version from %q", strings.TrimSpace(out))
215}