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