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