Federated via graft
8.7 kB
285 lines
1// Command sync runs the Forgejo <-> Radicle sync daemon: on an interval, it
2// mirrors git content, issues and patches for every repo pair listed in the
3// config file, then sleeps until the next pass.
4package main
5
6import (
7 "flag"
8 "fmt"
9 "log/slog"
10 "net/http"
11 "net/url"
12 "os"
13 "strings"
14 "time"
15
16 "graft/internal/activitypub"
17 "graft/internal/config"
18 "graft/internal/state"
19 "graft/internal/status"
20 "graft/internal/sync"
21)
22
23func main() {
24 configPath := flag.String("config", "/etc/graft/config.yaml", "path to config.yaml")
25 once := flag.Bool("once", false, "run a single pass and exit, instead of looping")
26 listen := flag.String("listen", ":8090", "address to serve /healthz and / (status) on; empty disables it")
27 flag.Parse()
28
29 log := slog.New(slog.NewJSONHandler(os.Stdout, nil))
30
31 cfg, err := config.Load(*configPath)
32 if err != nil {
33 log.Error("load config", "err", err)
34 os.Exit(1)
35 }
36
37 st, err := state.Open(cfg.StateDB)
38 if err != nil {
39 log.Error("open state db", "err", err)
40 os.Exit(1)
41 }
42 defer st.Close()
43
44 stateDir := cfg.StateDB
45 if idx := lastSlash(stateDir); idx >= 0 {
46 stateDir = stateDir[:idx]
47 }
48
49 syncers := make([]*sync.RepoSyncer, 0, len(cfg.Repos))
50 syncersByPair := map[string]*sync.RepoSyncer{}
51 syncersBySeries := map[string]*sync.RepoSyncer{}
52 pairsBySeries := map[string][]*sync.RepoSyncer{}
53 for _, pair := range cfg.Repos {
54 rs, err := sync.New(pair, st, sync.WorkDirFor(stateDir, pair.Name))
55 if err != nil {
56 log.Error("set up repo pair", "pair", pair.Name, "err", err)
57 os.Exit(1)
58 }
59 syncers = append(syncers, rs)
60 syncersByPair[pair.Name] = rs
61 if _, ok := syncersBySeries[rs.Series()]; !ok {
62 syncersBySeries[rs.Series()] = rs
63 }
64 pairsBySeries[rs.Series()] = append(pairsBySeries[rs.Series()], rs)
65 }
66 detectAuthorizedIntegrations(pairsBySeries, log)
67
68 // aiTargetHosts marks which series+Forgejo-host combinations receive
69 // pushes directly via Authorized Integrations rather than via graft
70 // itself — read by buildTopology so the dashboard can tell that side's
71 // silence apart from a side nothing has ever reached.
72 aiTargetHosts := map[string]map[string]bool{}
73 for series, pairs := range pairsBySeries {
74 for _, rs := range pairs {
75 if rs.HasAuthorizedIntegrationSource() {
76 if aiTargetHosts[series] == nil {
77 aiTargetHosts[series] = map[string]bool{}
78 }
79 aiTargetHosts[series][rs.ForgejoHost()] = true
80 }
81 }
82 }
83
84 topology := buildTopology(cfg, aiTargetHosts)
85 blueskyConfigured := map[string]bool{}
86 for _, pair := range cfg.Repos {
87 if pair.Bluesky != nil {
88 series := pair.Series
89 if series == "" {
90 series = pair.Name
91 }
92 blueskyConfigured[series] = true
93 }
94 }
95
96 tracker := status.NewTracker(st, cfg.SourceURL)
97 tracker.SetTopology(topology)
98 tracker.SetPublicHost(cfg.PublicHost)
99 tracker.SetBlueskyConfigured(blueskyConfigured)
100
101 var apHandler *activitypub.Handler
102 if cfg.PublicHost != "" {
103 apHandler = activitypub.NewHandler(st, cfg.PublicHost, log,
104 func(series string) string {
105 if refs := topology[series]; len(refs) > 0 {
106 return refs[0].URL
107 }
108 return ""
109 },
110 func(series string) bool {
111 _, ok := topology[series]
112 return ok
113 },
114 func(repoPair, kind string, forgejoID int64, radicleID, body string) error {
115 rs, ok := syncersByPair[repoPair]
116 if !ok {
117 return fmt.Errorf("unknown repo pair %q", repoPair)
118 }
119 return rs.CommentOnItem(kind, forgejoID, radicleID, body)
120 },
121 func(series string) string {
122 rs, ok := syncersBySeries[series]
123 if !ok {
124 return ""
125 }
126 did, err := rs.RadicleDID()
127 if err != nil {
128 log.Error("resolve radicle did", "series", series, "err", err)
129 return ""
130 }
131 return did
132 },
133 )
134 }
135
136 runAll := func() {
137 for _, rs := range syncers {
138 gitErr, issuesErr, patchErr := rs.Run(log)
139 tracker.Record(rs.Name(), rs.Series(), gitErr, issuesErr, patchErr)
140 }
141 if apHandler != nil {
142 apHandler.DeliverNewActivity()
143 }
144 }
145
146 if *listen != "" {
147 mux := http.NewServeMux()
148 if apHandler != nil {
149 apHandler.Register(mux)
150 }
151 mux.Handle("/", tracker.Handler())
152
153 go func() {
154 if err := http.ListenAndServe(*listen, securityHeaders(mux)); err != nil {
155 log.Error("status server stopped", "err", err)
156 }
157 }()
158 }
159
160 runAll()
161 if *once {
162 return
163 }
164
165 ticker := time.NewTicker(cfg.SyncInterval)
166 defer ticker.Stop()
167 for range ticker.C {
168 runAll()
169 }
170}
171
172// buildTopology declares every mirror target configured for each series
173// (its Forgejo side and its Radicle side), so the dashboard can show a
174// side that's never produced an event yet — not just sides the activity
175// log happens to mention.
176func buildTopology(cfg *config.Config, aiTargetHosts map[string]map[string]bool) map[string][]status.ServerRef {
177 topology := map[string][]status.ServerRef{}
178 seen := map[string]bool{}
179 add := func(series, host, repoURL string, radicle, ai bool) {
180 if host == "" {
181 return
182 }
183 key := series + "|" + host
184 if seen[key] {
185 return
186 }
187 seen[key] = true
188 topology[series] = append(topology[series], status.ServerRef{
189 Label: host,
190 URL: repoURL,
191 Radicle: radicle,
192 AuthorizedIntegration: ai,
193 })
194 }
195
196 for _, pair := range cfg.Repos {
197 series := pair.Series
198 if series == "" {
199 series = pair.Name
200 }
201
202 fHost := ""
203 if u, err := url.Parse(pair.Forgejo.BaseURL); err == nil {
204 fHost = u.Host
205 }
206 fURL := strings.TrimRight(pair.Forgejo.BaseURL, "/") + "/" + pair.Forgejo.Owner + "/" + pair.Forgejo.Repo
207 add(series, fHost, fURL, false, aiTargetHosts[series][fHost])
208
209 rHost := ""
210 if u, err := url.Parse(pair.Radicle.HTTPBaseURL); err == nil {
211 rHost = u.Host
212 }
213 add(series, rHost, sync.RadicleExplorerLink(pair.Radicle), true, false)
214 }
215 return topology
216}
217
218// detectAuthorizedIntegrations checks, for every pair of Forgejo hosts
219// sharing a series, whether one side already has a Forgejo Actions
220// workflow pushing directly into the other via Authorized Integrations
221// (see internal/sync's authint.go) — and if so, tells the receiving
222// pair's GitSyncer to stop relaying that content via Radicle itself,
223// since the direct path is faster. Run once at startup: workflow files
224// don't change often enough to justify re-checking every pass.
225func detectAuthorizedIntegrations(pairsBySeries map[string][]*sync.RepoSyncer, log *slog.Logger) {
226 for series, pairs := range pairsBySeries {
227 if len(pairs) < 2 {
228 continue
229 }
230 for _, source := range pairs {
231 for _, dest := range pairs {
232 if source == dest {
233 continue
234 }
235 destHost := dest.ForgejoHost()
236 if destHost == "" {
237 continue
238 }
239 if sync.HasAuthorizedIntegration(source.ForgejoClient(), destHost) {
240 log.Info("authorized integration detected",
241 "series", series, "source", source.ForgejoHost(), "target", destHost)
242 dest.SetAuthorizedIntegrationSource(source.ForgejoHost())
243 }
244 }
245 }
246 }
247}
248
249// securityHeaders sets a fixed, unconditional set of hardening response
250// headers on every request. Deliberately just static assignments — no
251// per-request logic, nothing that can panic or branch incorrectly — since
252// this wraps every route graft serves (dashboard, /social, ActivityPub,
253// /inbox) and a bug here would affect all of them at once.
254//
255// CSP is intentionally left out: the dashboard's entire stylesheet is one
256// inline <style> block, which a naive CSP would break outright (style-src
257// blocks inline styles by default) without a nonce wired through every
258// template render — worth doing properly later, not as a blanket header.
259// Cross-Origin-Embedder-Policy is left out too: it exists to protect
260// cross-origin-isolated contexts (SharedArrayBuffer, WASM threads) that
261// graft has no use for, and would only be a trap for a future contributor
262// who adds an external resource without thinking to CORS-enable it.
263func securityHeaders(next http.Handler) http.Handler {
264 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
265 h := w.Header()
266 h.Set("X-Content-Type-Options", "nosniff")
267 h.Set("X-Frame-Options", "DENY")
268 h.Set("Referrer-Policy", "strict-origin-when-cross-origin")
269 h.Set("Permissions-Policy", "camera=(), microphone=(), geolocation=(), payment=(), usb=(), interest-cohort=()")
270 h.Set("Cross-Origin-Opener-Policy", "same-origin")
271 // 180 days, no includeSubDomains (f1/f2/radicle aren't verified
272 // all-HTTPS as a unit) and no preload (effectively irreversible).
273 h.Set("Strict-Transport-Security", "max-age=15552000")
274 next.ServeHTTP(w, r)
275 })
276}
277
278func lastSlash(s string) int {
279 for i := len(s) - 1; i >= 0; i-- {
280 if s[i] == '/' {
281 return i
282 }
283 }
284 return -1
285}