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