Federated via graft
17 kB
534 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 "sort"
14 "strings"
15 "sync"
16 "time"
17
18 "graft/internal/activitypub"
19 "graft/internal/admin"
20 "graft/internal/config"
21 "graft/internal/state"
22 "graft/internal/status"
23 gsync "graft/internal/sync"
24)
25
26func main() {
27 configPath := flag.String("config", "/etc/graft/config.yaml", "path to config.yaml")
28 once := flag.Bool("once", false, "run a single pass and exit, instead of looping")
29 listen := flag.String("listen", ":8090", "address to serve /healthz and / (status) on; empty disables it")
30 flag.Parse()
31
32 log := slog.New(slog.NewJSONHandler(os.Stdout, nil))
33
34 cfg, err := config.Load(*configPath)
35 if err != nil {
36 log.Error("load config", "err", err)
37 os.Exit(1)
38 }
39 if cfg.AdminPassword == "graft" {
40 log.Warn("admin_password is left at its default value — anyone who knows it can add a peer or start a repo; set admin_password in config.yaml to change it")
41 }
42
43 st, err := state.Open(cfg.StateDB)
44 if err != nil {
45 log.Error("open state db", "err", err)
46 os.Exit(1)
47 }
48 defer st.Close()
49
50 stateDir := cfg.StateDB
51 if idx := lastSlash(stateDir); idx >= 0 {
52 stateDir = stateDir[:idx]
53 }
54 tokenDir := stateDir
55 if len(cfg.Repos) > 0 {
56 if idx := lastSlash(cfg.Repos[0].Forgejo.TokenFile); idx >= 0 {
57 tokenDir = cfg.Repos[0].Forgejo.TokenFile[:idx]
58 }
59 }
60 defaultRadHome := ""
61 if len(cfg.Repos) > 0 {
62 defaultRadHome = cfg.Repos[0].Radicle.RadHome
63 }
64
65 // live holds every piece of runtime state a dynamically-added peer or
66 // repo needs to join without a restart — mutated only from the
67 // runAll/ticker goroutine, but read concurrently by HTTP handler
68 // goroutines (ActivityPub callbacks, the admin package's Series
69 // lookup), so every access goes through mu.
70 live := &liveState{
71 pairs: append([]config.RepoPair(nil), cfg.Repos...),
72 syncersByPair: map[string]*gsync.RepoSyncer{},
73 syncersBySeries: map[string]*gsync.RepoSyncer{},
74 pairsBySeries: map[string][]*gsync.RepoSyncer{},
75 }
76 for _, pair := range cfg.Repos {
77 rs, err := gsync.New(pair, st, gsync.WorkDirFor(stateDir, pair.Name))
78 if err != nil {
79 log.Error("set up repo pair", "pair", pair.Name, "err", err)
80 os.Exit(1)
81 }
82 live.addSyncer(rs)
83 }
84 // dynamic_repo rows from a previous run persist in the database, but
85 // live.syncers starts empty every process start regardless of each
86 // row's "materialized" flag — that flag only means "graft has synced
87 // this at least once ever", not "this process already has a live
88 // RepoSyncer for it". Load all of them now, not just ones still
89 // pending their first sync.
90 loadAllDynamicRepos(st, live, stateDir, log)
91 detectAuthorizedIntegrations(live.pairsBySeriesSnapshot(), log)
92 live.rebuildTopology(cfg)
93
94 tracker := status.NewTracker(st, cfg.SourceURL)
95 tracker.SetTopology(live.topologySnapshot())
96 tracker.SetPublicHost(cfg.PublicHost)
97 tracker.SetBlueskyConfigured(blueskyConfigured(cfg.Repos))
98
99 var apHandler *activitypub.Handler
100 if cfg.PublicHost != "" {
101 apHandler = activitypub.NewHandler(st, cfg.PublicHost, log,
102 func(series string) string {
103 if refs := live.topologySnapshot()[series]; len(refs) > 0 {
104 return refs[0].URL
105 }
106 return ""
107 },
108 func(series string) bool {
109 _, ok := live.topologySnapshot()[series]
110 return ok
111 },
112 func(repoPair, kind string, forgejoID int64, radicleID, body string) error {
113 rs, ok := live.syncerForPair(repoPair)
114 if !ok {
115 return fmt.Errorf("unknown repo pair %q", repoPair)
116 }
117 return rs.CommentOnItem(kind, forgejoID, radicleID, body)
118 },
119 func(series string) string {
120 rs, ok := live.syncerForSeries(series)
121 if !ok {
122 return ""
123 }
124 did, err := rs.RadicleDID()
125 if err != nil {
126 log.Error("resolve radicle did", "series", series, "err", err)
127 return ""
128 }
129 return did
130 },
131 )
132 }
133
134 adminHandler := admin.NewHandler(admin.Config{
135 Store: st,
136 AdminPassword: cfg.AdminPassword,
137 PublicHost: cfg.PublicHost,
138 TokenDir: tokenDir,
139 DefaultRadHome: defaultRadHome,
140 Series: live.seriesInfos,
141 })
142
143 runAll := func() {
144 if materializeDynamicRepos(st, live, stateDir, log) {
145 detectAuthorizedIntegrations(live.pairsBySeriesSnapshot(), log)
146 live.rebuildTopology(cfg)
147 tracker.SetTopology(live.topologySnapshot())
148 tracker.SetBlueskyConfigured(blueskyConfigured(live.pairsSnapshot()))
149 }
150 for _, rs := range live.syncersSnapshot() {
151 gitErr, issuesErr, patchErr := rs.Run(log)
152 tracker.Record(rs.Name(), rs.Series(), gitErr, issuesErr, patchErr)
153 }
154 if apHandler != nil {
155 apHandler.DeliverNewActivity()
156 }
157 pollBlueskyReplies(live.pairsSnapshot(), live, st, log)
158 }
159
160 if *listen != "" {
161 mux := http.NewServeMux()
162 if apHandler != nil {
163 apHandler.Register(mux)
164 }
165 adminHandler.Register(mux)
166 mux.Handle("/", tracker.Handler())
167
168 go func() {
169 if err := http.ListenAndServe(*listen, securityHeaders(mux)); err != nil {
170 log.Error("status server stopped", "err", err)
171 }
172 }()
173 }
174
175 runAll()
176 if *once {
177 return
178 }
179
180 ticker := time.NewTicker(cfg.SyncInterval)
181 defer ticker.Stop()
182 for range ticker.C {
183 runAll()
184 }
185}
186
187// liveState is every piece of runtime registration that can grow after
188// startup (a peer or repo added through the dashboard), guarded by one
189// mutex since it's written from the ticker goroutine and read from HTTP
190// handler goroutines.
191type liveState struct {
192 mu sync.Mutex
193 pairs []config.RepoPair
194 syncers []*gsync.RepoSyncer
195 syncersByPair map[string]*gsync.RepoSyncer
196 syncersBySeries map[string]*gsync.RepoSyncer
197 pairsBySeries map[string][]*gsync.RepoSyncer
198 topology map[string][]status.ServerRef
199}
200
201func (l *liveState) addSyncer(rs *gsync.RepoSyncer) {
202 l.mu.Lock()
203 defer l.mu.Unlock()
204 l.syncers = append(l.syncers, rs)
205 l.syncersByPair[rs.Name()] = rs
206 if _, ok := l.syncersBySeries[rs.Series()]; !ok {
207 l.syncersBySeries[rs.Series()] = rs
208 }
209 l.pairsBySeries[rs.Series()] = append(l.pairsBySeries[rs.Series()], rs)
210}
211
212func (l *liveState) addPair(p config.RepoPair) {
213 l.mu.Lock()
214 defer l.mu.Unlock()
215 l.pairs = append(l.pairs, p)
216}
217
218func (l *liveState) syncersSnapshot() []*gsync.RepoSyncer {
219 l.mu.Lock()
220 defer l.mu.Unlock()
221 return append([]*gsync.RepoSyncer(nil), l.syncers...)
222}
223
224func (l *liveState) pairsSnapshot() []config.RepoPair {
225 l.mu.Lock()
226 defer l.mu.Unlock()
227 return append([]config.RepoPair(nil), l.pairs...)
228}
229
230func (l *liveState) pairsBySeriesSnapshot() map[string][]*gsync.RepoSyncer {
231 l.mu.Lock()
232 defer l.mu.Unlock()
233 out := make(map[string][]*gsync.RepoSyncer, len(l.pairsBySeries))
234 for k, v := range l.pairsBySeries {
235 out[k] = append([]*gsync.RepoSyncer(nil), v...)
236 }
237 return out
238}
239
240func (l *liveState) syncerForPair(name string) (*gsync.RepoSyncer, bool) {
241 l.mu.Lock()
242 defer l.mu.Unlock()
243 rs, ok := l.syncersByPair[name]
244 return rs, ok
245}
246
247func (l *liveState) syncerForSeries(series string) (*gsync.RepoSyncer, bool) {
248 l.mu.Lock()
249 defer l.mu.Unlock()
250 rs, ok := l.syncersBySeries[series]
251 return rs, ok
252}
253
254func (l *liveState) topologySnapshot() map[string][]status.ServerRef {
255 l.mu.Lock()
256 defer l.mu.Unlock()
257 return l.topology
258}
259
260// rebuildTopology recomputes the dashboard's topology map from the
261// current set of pairs (static config plus anything added dynamically)
262// and the AI-detection state each RepoSyncer already carries.
263func (l *liveState) rebuildTopology(cfg *config.Config) {
264 l.mu.Lock()
265 pairs := append([]config.RepoPair(nil), l.pairs...)
266 pairsBySeries := make(map[string][]*gsync.RepoSyncer, len(l.pairsBySeries))
267 for k, v := range l.pairsBySeries {
268 pairsBySeries[k] = v
269 }
270 l.mu.Unlock()
271
272 aiTargetHosts := map[string]map[string]bool{}
273 for series, rss := range pairsBySeries {
274 for _, rs := range rss {
275 if rs.HasAuthorizedIntegrationSource() {
276 if aiTargetHosts[series] == nil {
277 aiTargetHosts[series] = map[string]bool{}
278 }
279 aiTargetHosts[series][rs.ForgejoHost()] = true
280 }
281 }
282 }
283 topo := buildTopology(pairs, aiTargetHosts)
284
285 l.mu.Lock()
286 l.topology = topo
287 l.mu.Unlock()
288}
289
290// seriesInfos is the admin package's view of every federation a new peer
291// can join — one entry per series, carrying whichever pair's Radicle side
292// every sibling in that series already shares.
293func (l *liveState) seriesInfos() []admin.SeriesInfo {
294 pairs := l.pairsSnapshot()
295 seen := map[string]admin.SeriesInfo{}
296 var order []string
297 for _, p := range pairs {
298 series := p.Series
299 if series == "" {
300 series = p.Name
301 }
302 if _, ok := seen[series]; !ok {
303 seen[series] = admin.SeriesInfo{
304 Name: series,
305 RadicleRID: p.Radicle.RID,
306 RadicleHTTPBaseURL: p.Radicle.HTTPBaseURL,
307 RadicleExplorerURL: p.Radicle.ExplorerURL,
308 }
309 order = append(order, series)
310 }
311 }
312 sort.Strings(order)
313 out := make([]admin.SeriesInfo, 0, len(order))
314 for _, s := range order {
315 out = append(out, seen[s])
316 }
317 return out
318}
319
320func dynamicRepoToPair(d state.DynamicRepo) config.RepoPair {
321 return config.RepoPair{
322 Name: d.Name,
323 Series: d.Series,
324 Forgejo: config.ForgejoTarget{
325 BaseURL: d.ForgejoBaseURL, Owner: d.ForgejoOwner, Repo: d.ForgejoRepo, TokenFile: d.ForgejoTokenFile,
326 },
327 Radicle: config.RadicleTarget{
328 RID: d.RadicleRID, HTTPBaseURL: d.RadicleHTTPBaseURL, RadHome: d.RadicleRadHome, ExplorerURL: d.RadicleExplorerURL,
329 },
330 Sync: config.SyncScope{Git: d.SyncGit, Issues: d.SyncIssues, Patches: d.SyncPatches},
331 }
332}
333
334// loadAllDynamicRepos rebuilds live RepoSyncers for every dynamic_repo row
335// ever created, materialized or not — called once at startup, since
336// live.syncers is always empty at process start no matter what the
337// database's "materialized" flag says. Deliberately non-fatal (unlike a
338// bad static config.yaml pair, which does exit): the whole point of the
339// self-service onboarding path is resilience to transient failures (the
340// peer's Forgejo temporarily unreachable, say) — refusing to start graft
341// entirely over one dynamic peer would be a worse outcome than the bug
342// this function fixes. A row that fails here just stays out of the sync
343// loop until the next successful restart.
344func loadAllDynamicRepos(st *state.Store, live *liveState, stateDir string, log *slog.Logger) {
345 rows, err := st.AllDynamicRepos()
346 if err != nil {
347 log.Error("list dynamic repos", "err", err)
348 return
349 }
350 for _, d := range rows {
351 pair := dynamicRepoToPair(d)
352 rs, err := gsync.New(pair, st, gsync.WorkDirFor(stateDir, pair.Name))
353 if err != nil {
354 log.Error("load dynamic repo", "name", d.Name, "err", err)
355 continue
356 }
357 live.addSyncer(rs)
358 live.addPair(pair)
359 log.Info("loaded dynamic repo", "name", d.Name, "series", d.Series)
360 }
361}
362
363// materializeDynamicRepos turns every dynamic_repo row nobody has
364// activated yet into a live RepoSyncer, joining the sync loop on this
365// very pass — no restart, matching the "shows up next pass" model the
366// rest of graft already uses. Returns whether anything changed, so the
367// caller only pays for a topology rebuild when it's actually needed.
368func materializeDynamicRepos(st *state.Store, live *liveState, stateDir string, log *slog.Logger) bool {
369 rows, err := st.UnmaterializedDynamicRepos()
370 if err != nil {
371 log.Error("list dynamic repos", "err", err)
372 return false
373 }
374 changed := false
375 for _, d := range rows {
376 pair := dynamicRepoToPair(d)
377 rs, err := gsync.New(pair, st, gsync.WorkDirFor(stateDir, pair.Name))
378 if err != nil {
379 // Left unmaterialized — retried next pass rather than
380 // dropped, in case the failure is transient (e.g. the
381 // Forgejo repo wasn't reachable yet).
382 log.Error("materialize dynamic repo", "name", d.Name, "err", err)
383 continue
384 }
385 live.addSyncer(rs)
386 live.addPair(pair)
387 if err := st.MarkDynamicRepoMaterialized(d.ID); err != nil {
388 log.Error("mark dynamic repo materialized", "name", d.Name, "err", err)
389 }
390 log.Info("materialized dynamic repo", "name", d.Name, "series", d.Series)
391 changed = true
392 }
393 return changed
394}
395
396func blueskyConfigured(repos []config.RepoPair) map[string]bool {
397 out := map[string]bool{}
398 for _, pair := range repos {
399 if pair.Bluesky != nil {
400 series := pair.Series
401 if series == "" {
402 series = pair.Name
403 }
404 out[series] = true
405 }
406 }
407 return out
408}
409
410// buildTopology declares every mirror target configured for each series
411// (its Forgejo side and its Radicle side), so the dashboard can show a
412// side that's never produced an event yet — not just sides the activity
413// log happens to mention.
414func buildTopology(pairs []config.RepoPair, aiTargetHosts map[string]map[string]bool) map[string][]status.ServerRef {
415 topology := map[string][]status.ServerRef{}
416 seen := map[string]bool{}
417 add := func(series, host, repoURL, pairName string, radicle, ai bool) {
418 if host == "" {
419 return
420 }
421 key := series + "|" + host
422 if seen[key] {
423 return
424 }
425 seen[key] = true
426 topology[series] = append(topology[series], status.ServerRef{
427 Label: host,
428 URL: repoURL,
429 Radicle: radicle,
430 AuthorizedIntegration: ai,
431 PairName: pairName,
432 })
433 }
434
435 for _, pair := range pairs {
436 series := pair.Series
437 if series == "" {
438 series = pair.Name
439 }
440
441 fHost := ""
442 if u, err := url.Parse(pair.Forgejo.BaseURL); err == nil {
443 fHost = u.Host
444 }
445 fURL := strings.TrimRight(pair.Forgejo.BaseURL, "/") + "/" + pair.Forgejo.Owner + "/" + pair.Forgejo.Repo
446 add(series, fHost, fURL, pair.Name, false, aiTargetHosts[series][fHost])
447
448 rHost := ""
449 if u, err := url.Parse(pair.Radicle.HTTPBaseURL); err == nil {
450 rHost = u.Host
451 }
452 add(series, rHost, gsync.RadicleExplorerLink(pair.Radicle), pair.Name, true, false)
453 }
454 return topology
455}
456
457// detectAuthorizedIntegrations checks, for every pair of Forgejo hosts
458// sharing a series, whether one side already has a Forgejo Actions
459// workflow pushing directly into the other via Authorized Integrations
460// (see internal/sync's authint.go) — and if so, tells the receiving
461// pair's GitSyncer to stop relaying that content via Radicle itself,
462// since the direct path is faster. Run at startup and again whenever a
463// new peer joins dynamically — workflow files don't change often enough
464// to justify checking on every ordinary sync pass.
465func detectAuthorizedIntegrations(pairsBySeries map[string][]*gsync.RepoSyncer, log *slog.Logger) {
466 for series, pairs := range pairsBySeries {
467 if len(pairs) < 2 {
468 continue
469 }
470 for _, source := range pairs {
471 for _, dest := range pairs {
472 if source == dest {
473 continue
474 }
475 destHost := dest.ForgejoHost()
476 if destHost == "" {
477 continue
478 }
479 if gsync.HasAuthorizedIntegration(source.ForgejoClient(), destHost) {
480 log.Info("authorized integration detected",
481 "series", series, "source", source.ForgejoHost(), "target", destHost)
482 dest.SetAuthorizedIntegrationSource(source.ForgejoHost())
483 }
484 }
485 }
486 }
487}
488
489// securityHeaders sets a fixed, unconditional set of hardening response
490// headers on every request. Deliberately just static assignments — no
491// per-request logic, nothing that can panic or branch incorrectly — since
492// this wraps every route graft serves (dashboard, /social, ActivityPub,
493// /inbox, the admin onboarding forms) and a bug here would affect all of
494// them at once.
495//
496// Cross-Origin-Embedder-Policy is deliberately left out: it exists to
497// protect cross-origin-isolated contexts (SharedArrayBuffer, WASM threads)
498// that graft has no use for, and would only be a trap for a future
499// contributor who adds an external resource without thinking to
500// CORS-enable it.
501func securityHeaders(next http.Handler) http.Handler {
502 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
503 h := w.Header()
504 h.Set("X-Content-Type-Options", "nosniff")
505 h.Set("X-Frame-Options", "DENY")
506 h.Set("Referrer-Policy", "strict-origin-when-cross-origin")
507 h.Set("Permissions-Policy", "camera=(), microphone=(), geolocation=(), payment=(), usb=(), interest-cohort=()")
508 h.Set("Cross-Origin-Opener-Policy", "same-origin")
509 // 180 days, no includeSubDomains (f1/f2/radicle aren't verified
510 // all-HTTPS as a unit) and no preload (effectively irreversible).
511 h.Set("Strict-Transport-Security", "max-age=15552000")
512 // Tested locally (Playwright, real dashboard/social HTML, with a
513 // negative control proving the test actually catches breakage)
514 // before shipping: default-src 'none' blocks all script execution
515 // (the real XSS mitigation graft gets from this) without touching
516 // rendering, since style-src 'unsafe-inline' covers the one inline
517 // <style> block each page has — nothing else is ever injected into
518 // it, so there's no practical downside to allowing it specifically.
519 // form-action 'self': the admin onboarding/share pages are plain
520 // HTML forms posting back to graft itself — 'none' would silently
521 // block every one of them.
522 h.Set("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; img-src 'self'; connect-src 'self'; base-uri 'none'; form-action 'self'; frame-ancestors 'none'")
523 next.ServeHTTP(w, r)
524 })
525}
526
527func lastSlash(s string) int {
528 for i := len(s) - 1; i >= 0; i-- {
529 if s[i] == '/' {
530 return i
531 }
532 }
533 return -1
534}