graft.pds.cyberwild.org / graft-source
Federated via graft
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

Security audit fixes: SSRF hardening, replay protection, resource limits

Addresses every finding from the full-codebase security audit:

- CRITICAL: persistent SSRF via ActivityPub actor/inbox URLs. New
internal/activitypub/ssrf.go validates every externally-supplied URL
(https only, rejects loopback/private/link-local/metadata addresses)
and re-validates the resolved IP at dial time to close the DNS-rebinding
gap a resolve-then-connect check would leave open. Wired into
FetchActor, NewHandler's client, and DeliverNewActivity's per-follower
delivery loop.
- CRITICAL: state.db (holds every series' ActivityPub private key in
plaintext) was world-readable (644) on the live deployment. Open() now
hard-fails if it can't chmod the file to 600.
- HIGH: no replay protection on inbound HTTP Signatures. VerifyRequest
now rejects requests whose Date header is more than 5 minutes stale.
- MEDIUM: unbounded upstream response bodies embedded whole into error
messages/logs (Forgejo, AT Proto, ActivityPub clients) — capped at
4 MiB/1 MiB read limits with logged bodies truncated to 500 bytes.
- MEDIUM: subprocess calls (git, rad) had no timeout, so one hung command
could wedge the whole serial sync loop — bounded to 5 minutes via
context.WithTimeout/exec.CommandContext.
- MEDIUM: Authorized Integrations detection silently disables graft's own
push forever once triggered, with no bound — now falls back to pushing
directly after 10 consecutive skipped passes.

Also fixes a dashboard accuracy issue: a mirror side with zero
graft-logged events read as "nothing synced" even when it plainly held
the content — the Forgejo side changes originate on (graft only logs the
pushes it makes, never the pushes it reads from), or a side that received
content via a path graft doesn't log itself (Radicle's own peer-to-peer
gossip, or a Forgejo-to-Forgejo Authorized Integration). Those sides now
show a distinct "source" or "replicated" badge instead, with a legend
entry for each; "nothing synced" is reserved for a series with truly zero
activity anywhere. Also: renamed the dashboard subtitle to "Forgejos &
Radicles syncs", dropped the redundant "Activity by repo" heading, moved
the "last 3 days" window label to the Recent Activity header, and renamed
"Social bridges" to "Social IDs" throughout.

author
jeanbaptiste date (Sep 13, 2026, 8:58 AM +0200) commit 2add091b parent 277fa29c
+444 -46
+27 -6
cmd/sync/main.go
··· 65 65 } 66 66 detectAuthorizedIntegrations(pairsBySeries, log) 67 67 68 - topology := buildTopology(cfg) 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) 69 85 blueskyConfigured := map[string]bool{} 70 86 for _, pair := range cfg.Repos { 71 87 if pair.Bluesky != nil { ··· 157 173 // (its Forgejo side and its Radicle side), so the dashboard can show a 158 174 // side that's never produced an event yet — not just sides the activity 159 175 // log happens to mention. 160 - func buildTopology(cfg *config.Config) map[string][]status.ServerRef { 176 + func buildTopology(cfg *config.Config, aiTargetHosts map[string]map[string]bool) map[string][]status.ServerRef { 161 177 topology := map[string][]status.ServerRef{} 162 178 seen := map[string]bool{} 163 - add := func(series, host, repoURL string) { 179 + add := func(series, host, repoURL string, radicle, ai bool) { 164 180 if host == "" { 165 181 return 166 182 } ··· 169 185 return 170 186 } 171 187 seen[key] = true 172 - topology[series] = append(topology[series], status.ServerRef{Label: host, URL: repoURL}) 188 + topology[series] = append(topology[series], status.ServerRef{ 189 + Label: host, 190 + URL: repoURL, 191 + Radicle: radicle, 192 + AuthorizedIntegration: ai, 193 + }) 173 194 } 174 195 175 196 for _, pair := range cfg.Repos { ··· 183 204 fHost = u.Host 184 205 } 185 206 fURL := strings.TrimRight(pair.Forgejo.BaseURL, "/") + "/" + pair.Forgejo.Owner + "/" + pair.Forgejo.Repo 186 - add(series, fHost, fURL) 207 + add(series, fHost, fURL, false, aiTargetHosts[series][fHost]) 187 208 188 209 rHost := "" 189 210 if u, err := url.Parse(pair.Radicle.HTTPBaseURL); err == nil { 190 211 rHost = u.Host 191 212 } 192 - add(series, rHost, sync.RadicleExplorerLink(pair.Radicle)) 213 + add(series, rHost, sync.RadicleExplorerLink(pair.Radicle), true, false) 193 214 } 194 215 return topology 195 216 }
+9 -1
internal/activitypub/handlers.go
··· 57 57 known: known, 58 58 postComment: postComment, 59 59 radicleDID: radicleDID, 60 - client: &http.Client{Timeout: 15 * time.Second}, 60 + client: NewSafeClient(15 * time.Second), 61 61 } 62 62 } 63 63 ··· 398 398 continue 399 399 } 400 400 for _, f := range followers { 401 + // Defense in depth: FetchActor already validates an 402 + // inbox URL before a follower is ever stored, but check 403 + // again here too, in case older data predates that 404 + // validation or was written some other way. 405 + if err := ValidateExternalURL(f.InboxURI); err != nil { 406 + h.log.Error("ap delivery: refusing unsafe follower inbox", "series", series, "follower", f.ActorURI, "err", err) 407 + continue 408 + } 401 409 if err := PostSigned(h.client, f.InboxURI, keyID, priv, body); err != nil { 402 410 h.log.Error("ap delivery: post to follower", "series", series, "follower", f.ActorURI, "err", err) 403 411 }
+32 -2
internal/activitypub/httpsig.go
··· 42 42 return nil 43 43 } 44 44 45 + // maxClockSkew bounds how far a signed request's Date header may drift 46 + // from this server's clock before it's rejected — without this, a 47 + // captured valid signed request (Follow, reply, ...) could be replayed 48 + // indefinitely. 49 + const maxClockSkew = 5 * time.Minute 50 + 45 51 // VerifyRequest verifies an inbound request's Signature header against 46 52 // pub, re-deriving the signing string from whichever headers the 47 53 // signature itself claims to cover (a remote server may sign a different ··· 50 56 params := parseSignatureHeader(r.Header.Get("Signature")) 51 57 if params["signature"] == "" { 52 58 return fmt.Errorf("missing or unparseable Signature header") 59 + } 60 + 61 + if dateHeader := r.Header.Get("Date"); dateHeader != "" { 62 + signedAt, err := http.ParseTime(dateHeader) 63 + if err != nil { 64 + return fmt.Errorf("unparseable Date header: %w", err) 65 + } 66 + if skew := time.Since(signedAt); skew > maxClockSkew || skew < -maxClockSkew { 67 + return fmt.Errorf("Date header too far from current time (skew %s), possible replay", skew) 68 + } 69 + } else { 70 + return fmt.Errorf("missing Date header") 53 71 } 54 72 55 73 if digestHeader := r.Header.Get("Digest"); digestHeader != "" { ··· 134 152 } 135 153 defer resp.Body.Close() 136 154 if resp.StatusCode >= 300 { 137 - respBody, _ := io.ReadAll(resp.Body) 138 - return fmt.Errorf("post %s: HTTP %d: %s", url, resp.StatusCode, string(respBody)) 155 + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, maxResponseBody)) 156 + return fmt.Errorf("post %s: HTTP %d: %s", url, resp.StatusCode, truncateForLog(respBody)) 139 157 } 140 158 return nil 141 159 } 160 + 161 + // truncateForLog caps how much of a remote error body ends up embedded in 162 + // an error message (and from there, in logs) — a remote server's response 163 + // is untrusted content, and shouldn't get an unbounded write into our own 164 + // logging. 165 + func truncateForLog(b []byte) string { 166 + const max = 500 167 + if len(b) <= max { 168 + return string(b) 169 + } 170 + return string(b[:max]) + "... (truncated)" 171 + }
+43
internal/activitypub/httpsig_test.go
··· 2 2 3 3 import ( 4 4 "bytes" 5 + "crypto" 6 + "crypto/rand" 5 7 "crypto/rsa" 8 + "crypto/sha256" 9 + "encoding/base64" 10 + "fmt" 6 11 "io" 7 12 "net/http" 8 13 "net/http/httptest" 14 + "strings" 9 15 "testing" 16 + "time" 10 17 ) 11 18 12 19 // TestSignVerifyRoundTrip exercises the actual code path a remote server ··· 64 71 tampered := []byte(`{"type":"Follow","actor":"https://attacker.example/users/mallory"}`) 65 72 if err := VerifyRequest(req, tampered, pub); err == nil { 66 73 t.Fatal("expected verification to fail on tampered body, got nil error") 74 + } 75 + } 76 + 77 + // TestVerifyRejectsStaleDate confirms a byte-for-byte replay of an old, 78 + // genuinely-validly-signed request is rejected: the signature itself 79 + // verifies fine (Date is part of what was signed, and hasn't changed — 80 + // this is a faithful replay, not tampering), but the freshness check must 81 + // still catch that "now" is far past the signed Date. SignRequest always 82 + // stamps the current time, so this builds the signature by hand with a 83 + // fixed old Date to simulate a captured-and-replayed request correctly. 84 + func TestVerifyRejectsStaleDate(t *testing.T) { 85 + _, pub, priv := testKeyPair(t) 86 + body := []byte(`{"type":"Follow","actor":"https://example.social/users/alice"}`) 87 + keyID := "https://example.social/users/alice#main-key" 88 + 89 + req, err := http.NewRequest(http.MethodPost, "https://graft.example.org/actors/testrepo/inbox", bytes.NewReader(body)) 90 + if err != nil { 91 + t.Fatal(err) 92 + } 93 + oldDate := time.Now().Add(-1 * time.Hour).UTC().Format(http.TimeFormat) 94 + digest := sha256.Sum256(body) 95 + req.Header.Set("Digest", "SHA-256="+base64.StdEncoding.EncodeToString(digest[:])) 96 + req.Header.Set("Date", oldDate) 97 + 98 + signingString := buildSigningString(signedHeaders, req.Method, req.URL.RequestURI(), req.URL.Host, req.Header) 99 + hashed := sha256.Sum256([]byte(signingString)) 100 + sig, err := rsa.SignPKCS1v15(rand.Reader, priv, crypto.SHA256, hashed[:]) 101 + if err != nil { 102 + t.Fatal(err) 103 + } 104 + req.Header.Set("Signature", fmt.Sprintf( 105 + `keyId="%s",algorithm="rsa-sha256",headers="%s",signature="%s"`, 106 + keyID, strings.Join(signedHeaders, " "), base64.StdEncoding.EncodeToString(sig))) 107 + 108 + if err := VerifyRequest(req, body, pub); err == nil { 109 + t.Fatal("expected verification to fail on a stale/replayed Date header, got nil error") 67 110 } 68 111 } 69 112
+16 -6
internal/activitypub/remote.go
··· 3 3 import ( 4 4 "encoding/json" 5 5 "fmt" 6 + "io" 6 7 "net/http" 7 8 ) 8 9 9 - // FetchActor retrieves a remote actor object. An unsigned GET is used — 10 - // acceptable here since this is only for resolving a Follow request's 11 - // actor (to get their public key and inbox URL) and most servers permit 12 - // unauthenticated actor fetches; authenticated fetches ("secure mode") are 13 - // out of scope for now. 10 + // FetchActor retrieves a remote actor object. actorURL is attacker 11 + // influenced (it comes straight off an inbound activity's "actor" field), 12 + // so it's validated before any network call, and client is expected to be 13 + // one from NewSafeClient. An unsigned GET is used — acceptable here since 14 + // this is only for resolving a Follow request's actor (to get their 15 + // public key and inbox URL) and most servers permit unauthenticated actor 16 + // fetches; authenticated fetches ("secure mode") are out of scope for now. 14 17 func FetchActor(client *http.Client, actorURL string) (*Actor, error) { 18 + if err := ValidateExternalURL(actorURL); err != nil { 19 + return nil, fmt.Errorf("actor URL: %w", err) 20 + } 21 + 15 22 req, err := http.NewRequest(http.MethodGet, actorURL, nil) 16 23 if err != nil { 17 24 return nil, fmt.Errorf("build request: %w", err) ··· 28 35 } 29 36 30 37 var a Actor 31 - if err := json.NewDecoder(resp.Body).Decode(&a); err != nil { 38 + if err := json.NewDecoder(io.LimitReader(resp.Body, maxResponseBody)).Decode(&a); err != nil { 32 39 return nil, fmt.Errorf("decode actor %s: %w", actorURL, err) 40 + } 41 + if err := ValidateExternalURL(a.Inbox); err != nil { 42 + return nil, fmt.Errorf("actor inbox URL: %w", err) 33 43 } 34 44 return &a, nil 35 45 }
+91
internal/activitypub/ssrf.go
··· 1 + package activitypub 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "net" 7 + "net/http" 8 + "net/url" 9 + "time" 10 + ) 11 + 12 + // maxResponseBody caps how much of a remote response this package will 13 + // ever read into memory — a hostile or misbehaving server (notably one 14 + // controlled by whoever sent us a Follow) gets cut off rather than able to 15 + // exhaust memory. 16 + const maxResponseBody = 1 << 20 // 1 MiB 17 + 18 + // ValidateExternalURL rejects non-https URLs and the most obvious 19 + // local/loopback hostnames before any network call is made. A cheap first 20 + // check ahead of the dialer-level one NewSafeClient performs — every 21 + // outbound target here (a Follow's actor, an actor's inbox URL, ...) is 22 + // attacker-influenced, so both checks matter: this one for a fast, clear 23 + // rejection; the dialer check to close the DNS-rebinding gap a 24 + // resolve-then-separately-connect check would otherwise leave open. 25 + func ValidateExternalURL(rawURL string) error { 26 + u, err := url.Parse(rawURL) 27 + if err != nil { 28 + return fmt.Errorf("invalid URL: %w", err) 29 + } 30 + if u.Scheme != "https" { 31 + return fmt.Errorf("only https URLs are allowed, got %q", u.Scheme) 32 + } 33 + host := u.Hostname() 34 + if host == "" { 35 + return fmt.Errorf("URL has no host") 36 + } 37 + if host == "localhost" { 38 + return fmt.Errorf("refusing to connect to localhost") 39 + } 40 + if ip := net.ParseIP(host); ip != nil && isUnsafeIP(ip) { 41 + return fmt.Errorf("refusing to connect to disallowed address %s", ip) 42 + } 43 + return nil 44 + } 45 + 46 + // NewSafeClient returns an http.Client hardened for outbound requests to 47 + // attacker-influenced URLs: no redirects are followed (a validated URL 48 + // could otherwise redirect somewhere unvalidated), and every connection's 49 + // actually-resolved IP is checked against private/loopback/link-local 50 + // ranges immediately before dialing it — not just at an earlier 51 + // validation step, which a DNS answer that changes between check and 52 + // connect could bypass. 53 + func NewSafeClient(timeout time.Duration) *http.Client { 54 + dialer := &net.Dialer{Timeout: 10 * time.Second} 55 + transport := &http.Transport{ 56 + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { 57 + host, port, err := net.SplitHostPort(addr) 58 + if err != nil { 59 + return nil, err 60 + } 61 + ips, err := net.DefaultResolver.LookupIP(ctx, "ip", host) 62 + if err != nil { 63 + return nil, fmt.Errorf("resolve %s: %w", host, err) 64 + } 65 + var safe net.IP 66 + for _, ip := range ips { 67 + if isUnsafeIP(ip) { 68 + continue 69 + } 70 + safe = ip 71 + break 72 + } 73 + if safe == nil { 74 + return nil, fmt.Errorf("refusing to connect to %s: no public address resolved", host) 75 + } 76 + return dialer.DialContext(ctx, network, net.JoinHostPort(safe.String(), port)) 77 + }, 78 + } 79 + return &http.Client{ 80 + Timeout: timeout, 81 + Transport: transport, 82 + CheckRedirect: func(req *http.Request, via []*http.Request) error { 83 + return http.ErrUseLastResponse 84 + }, 85 + } 86 + } 87 + 88 + func isUnsafeIP(ip net.IP) bool { 89 + return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || 90 + ip.IsLinkLocalMulticast() || ip.IsUnspecified() || ip.IsMulticast() 91 + }
+39
internal/activitypub/ssrf_test.go
··· 1 + package activitypub 2 + 3 + import "testing" 4 + 5 + func TestValidateExternalURLRejectsUnsafeTargets(t *testing.T) { 6 + cases := []struct { 7 + name string 8 + url string 9 + }{ 10 + {"plain http", "http://example.social/users/alice"}, 11 + {"loopback IP", "https://127.0.0.1/actors/x"}, 12 + {"IPv6 loopback", "https://[::1]/actors/x"}, 13 + {"localhost", "https://localhost/actors/x"}, 14 + {"private range 10.x", "https://10.0.0.5/actors/x"}, 15 + {"private range 192.168.x", "https://192.168.1.1/actors/x"}, 16 + {"cloud metadata address", "https://169.254.169.254/latest/meta-data/"}, 17 + {"no host", "https:///actors/x"}, 18 + {"not a URL", "not a url at all"}, 19 + } 20 + for _, tc := range cases { 21 + t.Run(tc.name, func(t *testing.T) { 22 + if err := ValidateExternalURL(tc.url); err == nil { 23 + t.Fatalf("expected %q to be rejected, got nil error", tc.url) 24 + } 25 + }) 26 + } 27 + } 28 + 29 + func TestValidateExternalURLAllowsPublicHTTPS(t *testing.T) { 30 + cases := []string{ 31 + "https://example.social/users/alice", 32 + "https://mastodon.social/users/alice#main-key", 33 + } 34 + for _, u := range cases { 35 + if err := ValidateExternalURL(u); err != nil { 36 + t.Errorf("expected %q to be allowed, got error: %v", u, err) 37 + } 38 + } 39 + }
+21 -5
internal/atproto/client.go
··· 21 21 // for graft's short auto-generated notes). 22 22 const maxPostRunes = 300 23 23 24 + // maxResponseBody caps how much of a PDS response this client reads into 25 + // memory; maxLoggedBody further caps how much of an error body ends up 26 + // embedded in an error message (and from there, in logs) — untrusted 27 + // upstream content shouldn't get an unbounded write into our own logging. 28 + const ( 29 + maxResponseBody = 1 << 20 // 1 MiB 30 + maxLoggedBody = 500 31 + ) 32 + 33 + func truncate(b []byte) string { 34 + if len(b) <= maxLoggedBody { 35 + return string(b) 36 + } 37 + return string(b[:maxLoggedBody]) + "... (truncated)" 38 + } 39 + 24 40 // Client is a session against one AT Proto PDS, authenticated as one 25 41 // account. 26 42 type Client struct { ··· 61 77 } 62 78 defer resp.Body.Close() 63 79 if resp.StatusCode >= 300 { 64 - b, _ := io.ReadAll(resp.Body) 65 - return fmt.Errorf("create session: HTTP %d: %s", resp.StatusCode, b) 80 + b, _ := io.ReadAll(io.LimitReader(resp.Body, maxResponseBody)) 81 + return fmt.Errorf("create session: HTTP %d: %s", resp.StatusCode, truncate(b)) 66 82 } 67 83 68 84 var out struct { 69 85 DID string `json:"did"` 70 86 AccessJwt string `json:"accessJwt"` 71 87 } 72 - if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { 88 + if err := json.NewDecoder(io.LimitReader(resp.Body, maxResponseBody)).Decode(&out); err != nil { 73 89 return fmt.Errorf("decode session: %w", err) 74 90 } 75 91 c.did = out.DID ··· 115 131 } 116 132 defer resp.Body.Close() 117 133 if resp.StatusCode >= 300 { 118 - b, _ := io.ReadAll(resp.Body) 119 - return fmt.Errorf("create record: HTTP %d: %s", resp.StatusCode, b) 134 + b, _ := io.ReadAll(io.LimitReader(resp.Body, maxResponseBody)) 135 + return fmt.Errorf("create record: HTTP %d: %s", resp.StatusCode, truncate(b)) 120 136 } 121 137 return nil 122 138 }
+20 -3
internal/forgejo/client.go
··· 14 14 "time" 15 15 ) 16 16 17 + // maxResponseBody caps how much of a Forgejo response this client reads 18 + // into memory. 19 + const maxResponseBody = 4 << 20 // 4 MiB 20 + 21 + // maxLoggedBody caps how much of an error response body gets embedded in 22 + // an error message (and from there, into logs) — an upstream server's 23 + // response is content we don't control, so it shouldn't get an unbounded 24 + // write into our own logging. 25 + const maxLoggedBody = 500 26 + 17 27 // Client talks to one Forgejo instance on behalf of one repository. 18 28 type Client struct { 19 29 baseURL string ··· 58 68 defer resp.Body.Close() 59 69 60 70 if resp.StatusCode >= 300 { 61 - respBody, _ := io.ReadAll(resp.Body) 62 - return fmt.Errorf("%s %s: HTTP %d: %s", method, path, resp.StatusCode, string(respBody)) 71 + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, maxResponseBody)) 72 + return fmt.Errorf("%s %s: HTTP %d: %s", method, path, resp.StatusCode, truncate(respBody)) 63 73 } 64 74 65 75 if out != nil { 66 - if err := json.NewDecoder(resp.Body).Decode(out); err != nil { 76 + if err := json.NewDecoder(io.LimitReader(resp.Body, maxResponseBody)).Decode(out); err != nil { 67 77 return fmt.Errorf("decode response for %s %s: %w", method, path, err) 68 78 } 69 79 } 70 80 return nil 81 + } 82 + 83 + func truncate(b []byte) string { 84 + if len(b) <= maxLoggedBody { 85 + return string(b) 86 + } 87 + return string(b[:maxLoggedBody]) + "... (truncated)" 71 88 } 72 89 73 90 // Repository is the subset of Forgejo's repository object the daemon uses.
+15 -2
internal/radicle/client.go
··· 6 6 7 7 import ( 8 8 "bytes" 9 + "context" 9 10 "encoding/json" 10 11 "fmt" 12 + "io" 11 13 "net/http" 12 14 "os" 13 15 "os/exec" 14 16 "regexp" 15 17 "time" 16 18 ) 19 + 20 + // execTimeout bounds every "rad" CLI subprocess — see the matching 21 + // constant and comment in internal/sync/git.go for why this matters: a 22 + // hung subprocess would otherwise block the whole sync loop. 23 + const execTimeout = 5 * time.Minute 24 + 25 + // maxResponseBody caps how much of a radicle-httpd response this client 26 + // reads into memory. 27 + const maxResponseBody = 4 << 20 // 4 MiB — patch/issue listings can be large 17 28 18 29 // Client operates on one repository (identified by its RID) tracked by one 19 30 // local Radicle node. ··· 44 55 if resp.StatusCode >= 300 { 45 56 return fmt.Errorf("GET %s: HTTP %d", path, resp.StatusCode) 46 57 } 47 - return json.NewDecoder(resp.Body).Decode(out) 58 + return json.NewDecoder(io.LimitReader(resp.Body, maxResponseBody)).Decode(out) 48 59 } 49 60 50 61 // rad runs the CLI against this client's repository and RAD_HOME, returning 51 62 // combined stdout+stderr for logging on error. 52 63 func (c *Client) rad(args ...string) (string, error) { 53 64 args = append(args, "--repo", c.rid) 54 - cmd := exec.Command(c.radBin, args...) 65 + ctx, cancel := context.WithTimeout(context.Background(), execTimeout) 66 + defer cancel() 67 + cmd := exec.CommandContext(ctx, c.radBin, args...) 55 68 cmd.Env = append(os.Environ(), "RAD_HOME="+c.radHome) 56 69 var out bytes.Buffer 57 70 cmd.Stdout = &out
+9
internal/state/store.go
··· 6 6 import ( 7 7 "database/sql" 8 8 "fmt" 9 + "os" 9 10 "strings" 10 11 "time" 11 12 ··· 26 27 if err := migrate(db); err != nil { 27 28 db.Close() 28 29 return nil, fmt.Errorf("migrate state db: %w", err) 30 + } 31 + // This database holds every series' ActivityPub private key 32 + // (ap_actor.private_key) in plaintext — it must never be 33 + // group/world-readable, regardless of the umask the daemon happened to 34 + // start with. Best-effort: sqlite may not have created the file yet on 35 + // the very first Open before migrate runs, but by this point it has. 36 + if err := os.Chmod(path, 0o600); err != nil { 37 + return nil, fmt.Errorf("restrict state db permissions: %w", err) 29 38 } 30 39 return &Store{db: db}, nil 31 40 }
+77 -15
internal/status/status.go
··· 36 36 // instance or a specific Radicle node — as configured, regardless of 37 37 // whether it has ever actually produced an event yet. 38 38 type ServerRef struct { 39 - Label string // host, e.g. "f1.cyberwild.org" 40 - URL string // repo root on that host 39 + Label string // host, e.g. "f1.cyberwild.org" 40 + URL string // repo root on that host 41 + Radicle bool // true if this side is a Radicle node 42 + AuthorizedIntegration bool // true if this Forgejo side receives pushes directly from another pair's Forgejo, bypassing graft 41 43 } 42 44 43 45 // Tracker holds the latest result per repo pair, safe for concurrent use, ··· 305 307 // with 3 mirrored sides (two Forgejos plus Radicle) gets 3 sub-rows, never 306 308 // one link picked arbitrarily to represent all of them. 307 309 type subRow struct { 308 - Label string 309 - URL string 310 - Events []event 310 + Label string 311 + URL string 312 + // EmptyState explains why this side shows no events even though it may 313 + // already hold the mirrored content — empty when Events is non-empty, 314 + // or when the whole series has genuinely never been touched. "source" 315 + // marks the Forgejo side changes originate on (graft only logs the 316 + // pushes it makes, never the pushes it reads from); "replicated" marks 317 + // a side that received the content some other way graft doesn't log 318 + // itself — Radicle's own peer-to-peer gossip between nodes, or a 319 + // Forgejo-to-Forgejo Authorized Integration. 320 + EmptyState string 321 + Events []event 311 322 } 312 323 313 324 // event is one activity_log row: one heatmap cell, one tooltip, one link. ··· 373 384 // behind one shared tooltip. 374 385 func buildSeriesRows(entries []state.ActivityEntry, health map[string]bool, topology map[string][]ServerRef) []seriesRow { 375 386 type subAcc struct { 376 - url string 377 - events []state.ActivityEntry 387 + url string 388 + events []state.ActivityEntry 389 + radicle bool 390 + ai bool 378 391 } 379 392 type seriesAcc struct { 380 393 subOrder []string ··· 412 425 if sub.url == "" { 413 426 sub.url = ref.URL 414 427 } 428 + sub.radicle = ref.Radicle 429 + sub.ai = ref.AuthorizedIntegration 415 430 } 416 431 } 417 432 ··· 442 457 subOrder := append([]string(nil), sa.subOrder...) 443 458 sort.Strings(subOrder) 444 459 460 + // A series with at least one logged event anywhere means every 461 + // currently-silent side already holds the mirrored content too — 462 + // either as the origin, or received via a path graft doesn't log 463 + // itself (Radicle gossip, an Authorized Integration). Only a 464 + // series with zero events anywhere is genuinely untouched. 465 + anyEvents := false 466 + for _, label := range subOrder { 467 + if len(sa.subs[label].events) > 0 { 468 + anyEvents = true 469 + break 470 + } 471 + } 472 + 445 473 subRows := make([]subRow, 0, len(subOrder)) 446 474 for _, label := range subOrder { 447 475 sub := sa.subs[label] ··· 450 478 es := make([]state.ActivityEntry, len(sub.events)) 451 479 copy(es, sub.events) 452 480 sort.Slice(es, func(i, j int) bool { return es[i].OccurredAt.Before(es[j].OccurredAt) }) 453 - subRows = append(subRows, subRow{Label: label, URL: sub.url, Events: toEvents(es)}) 481 + 482 + emptyState := "" 483 + if len(es) == 0 && anyEvents { 484 + switch { 485 + case sub.ai, sub.radicle: 486 + emptyState = "replicated" 487 + default: 488 + emptyState = "source" 489 + } 490 + } 491 + subRows = append(subRows, subRow{Label: label, URL: sub.url, EmptyState: emptyState, Events: toEvents(es)}) 454 492 } 455 493 456 494 ok, known := health[name] ··· 564 602 --kind-git: #0052cc; 565 603 --kind-issue: #00875a; 566 604 --kind-patch: #6554c0; 605 + --kind-source: #ff991f; 606 + --kind-replicated: #00b8d9; 567 607 --bad: #de350b; 568 608 --good: #36b37e; 569 609 } ··· 594 634 .dot { width: 7px; height: 7px; border-radius: 50%; background: var(--good); flex: none; display: inline-block; } 595 635 .dot.bad { background: var(--bad); } 596 636 597 - .heatmap-head { display: flex; align-items: baseline; justify-content: space-between; } 637 + .heatmap-head { display: flex; align-items: baseline; justify-content: space-between; margin-bottom: .9rem; } 598 638 .heatmap-head h2 { margin: 0; } 599 639 .legend { display: flex; gap: .9rem; margin-bottom: .9rem; } 600 640 .legend span { display: flex; align-items: center; gap: .35rem; font-size: .74rem; color: var(--text-muted); } ··· 602 642 .legend .sw[data-kind="git"] { background: var(--kind-git); } 603 643 .legend .sw[data-kind="issue"] { background: var(--kind-issue); } 604 644 .legend .sw[data-kind="patch"] { background: var(--kind-patch); } 645 + .legend .sw[data-kind="source"] { background: var(--kind-source); } 646 + .legend .sw[data-kind="replicated"] { background: var(--kind-replicated); } 605 647 .heatmap { display: flex; flex-direction: column; gap: .15rem; } 606 648 .series-group { border-bottom: 1px solid var(--border); } 607 649 .series-group:last-child { border-bottom: none; } ··· 656 698 .tip .tip-kind { color: #b3bac5; flex: none; } 657 699 .tip .tip-empty { padding: 0 .75rem; color: #b3bac5; } 658 700 701 + .state-pill { 702 + display: inline-flex; align-items: center; gap: .35rem; 703 + font-size: .74rem; padding: .2rem .6rem; border-radius: 999px; 704 + } 705 + .state-pill .state-dot { width: 7px; height: 7px; border-radius: 50%; flex: none; } 706 + .state-pill.state-source { background: rgba(255,153,31,.14); color: #974f0c; } 707 + .state-pill.state-source .state-dot { background: var(--kind-source); } 708 + .state-pill.state-replicated { background: rgba(0,184,217,.14); color: #0868a0; } 709 + .state-pill.state-replicated .state-dot { background: var(--kind-replicated); } 710 + 659 711 h2 { font-size: .78rem; text-transform: uppercase; letter-spacing: .04em; color: var(--text-muted); margin: 0 0 .9rem; font-weight: 600; } 660 712 .commits { border-top: 1px solid var(--border); } 661 713 .commit { ··· 691 743 <div class="top"> 692 744 <div> 693 745 <h1>graft</h1> 694 - <span class="sub">Forgejo &lt;-&gt; Radicle sync — last 3 days</span> 746 + <span class="sub">Forgejos &amp; Radicles syncs</span> 695 747 </div> 696 - {{if .SocialEnabled}}<a class="nav-link" href="/social">Social bridges &rarr;</a>{{end}} 748 + {{if .SocialEnabled}}<a class="nav-link" href="/social">Social IDs &rarr;</a>{{end}} 697 749 </div> 698 750 699 751 <div class="card"> 700 752 <div class="heatmap-head"> 701 - <h2>Activity by repo</h2> 702 753 <div class="legend"> 703 754 <span><i class="sw" data-kind="git"></i>Commit</span> 704 755 <span><i class="sw" data-kind="issue"></i>Issue</span> 705 756 <span><i class="sw" data-kind="patch"></i>Patch</span> 757 + <span><i class="sw" data-kind="source"></i>Source</span> 758 + <span><i class="sw" data-kind="replicated"></i>Replicated</span> 706 759 </div> 707 760 </div> 708 761 <div class="heatmap"> ··· 734 787 </div> 735 788 </div> 736 789 {{else}} 790 + {{if eq .EmptyState "source"}} 791 + <span class="state-pill state-source" title="content originates here — graft only logs the pushes it makes, not the pushes it reads from"><span class="state-dot"></span>source</span> 792 + {{else if eq .EmptyState "replicated"}} 793 + <span class="state-pill state-replicated" title="content arrived here without a graft-logged push — Radicle peer-to-peer gossip, or a Forgejo Authorized Integration"><span class="state-dot"></span>replicated</span> 794 + {{else}} 737 795 <span class="tip-empty">nothing synced</span> 796 + {{end}} 738 797 {{end}} 739 798 </div> 740 799 </div> ··· 748 807 </div> 749 808 750 809 <div class="card"> 751 - <h2>Recent activity</h2> 810 + <div class="heatmap-head"> 811 + <h2>Recent activity</h2> 812 + <span class="sub">last 3 days</span> 813 + </div> 752 814 <div class="commits"> 753 815 {{range .Activity}} 754 816 <div class="commit"> ··· 784 846 <head> 785 847 <meta charset="utf-8"> 786 848 <meta name="viewport" content="width=device-width, initial-scale=1"> 787 - <title>graft — social bridges</title> 849 + <title>graft — social IDs</title> 788 850 <style> 789 851 :root { 790 852 color-scheme: light; ··· 834 896 <main> 835 897 <div class="top"> 836 898 <div> 837 - <h1>graft — social bridges</h1> 899 + <h1>graft — social IDs</h1> 838 900 <span class="sub">ActivityPub + AT Proto, per repo</span> 839 901 </div> 840 902 <a class="nav-link" href="/">&larr; Dashboard</a>
+32 -5
internal/sync/git.go
··· 11 11 12 12 import ( 13 13 "bytes" 14 + "context" 14 15 "fmt" 15 16 "os" 16 17 "os/exec" 17 18 "path/filepath" 18 19 "strings" 20 + "time" 19 21 20 22 "graft/internal/state" 21 23 ) 24 + 25 + // execTimeout bounds every git/rad subprocess this package runs. Without 26 + // it, one hung fetch/push (a slow network, an unresponsive Radicle node) 27 + // would block the single-threaded sync loop forever — every other 28 + // configured pair too, not just the stuck one. 29 + const execTimeout = 5 * time.Minute 22 30 23 31 // GitSyncer mirrors a repository's default branch between Forgejo and 24 32 // Radicle using a local working clone with two remotes. ··· 41 49 // this pair waiting for the next poll to relay it via Radicle. When 42 50 // set, Sync skips its own rad -> forgejo push and just observes 43 51 // whether the content has already arrived, falling back to pushing it 44 - // itself only once a real divergence would otherwise be reported. 52 + // itself only once a real divergence would otherwise be reported — 53 + // or after maxAISkipStreak consecutive passes without convergence 54 + // (see Sync), since detection here is a content heuristic (authint.go), 55 + // not a guarantee the workflow is actually delivering. 45 56 AuthorizedIntegrationSource string 57 + aiSkipStreak int 46 58 } 47 59 60 + // maxAISkipStreak bounds how many consecutive passes GitSyncer will trust 61 + // AuthorizedIntegrationSource without seeing forgejo actually catch up to 62 + // radicle. Detection is a heuristic (a matching string in a workflow 63 + // file), not proof the integration is configured correctly or even still 64 + // working — without this bound, a stale or misconfigured workflow would 65 + // make graft silently stop mirroring that direction forever. 66 + const maxAISkipStreak = 10 67 + 48 68 func (g *GitSyncer) run(args ...string) (string, error) { 49 - cmd := exec.Command("git", args...) 69 + ctx, cancel := context.WithTimeout(context.Background(), execTimeout) 70 + defer cancel() 71 + cmd := exec.CommandContext(ctx, "git", args...) 50 72 cmd.Dir = g.WorkDir 51 73 cmd.Env = append(os.Environ(), "RAD_HOME="+g.RadHome) 52 74 var out bytes.Buffer ··· 98 120 // --nid is deprecated in favor of `rad node status --only nid` and 99 121 // prints a warning to stderr; keep that separate from stdout so it 100 122 // never ends up concatenated into the id itself. 101 - cmd := exec.Command("rad", "self", "--nid") 123 + ctx, cancel := context.WithTimeout(context.Background(), execTimeout) 124 + defer cancel() 125 + cmd := exec.CommandContext(ctx, "rad", "self", "--nid") 102 126 cmd.Env = append(os.Environ(), "RAD_HOME="+g.RadHome) 103 127 var out, errOut bytes.Buffer 104 128 cmd.Stdout = &out ··· 158 182 case forgejoHead == radHead: 159 183 // Already in sync; still record the cursor in case this is the 160 184 // first run and it was empty before. 185 + g.aiSkipStreak = 0 161 186 162 187 case radHead == "" || radHead == lastRad: 163 188 // Radicle hasn't moved since we last looked: forgejo is ahead. ··· 167 192 168 193 case forgejoHead == "" || forgejoHead == lastForgejo: 169 194 // Forgejo hasn't moved: rad is ahead. 170 - if g.AuthorizedIntegrationSource != "" { 195 + if g.AuthorizedIntegrationSource != "" && g.aiSkipStreak < maxAISkipStreak { 171 196 // Someone else's CI already pushes directly into this 172 197 // Forgejo (see AuthorizedIntegrationSource's doc comment) — 173 198 // faster than waiting for us to relay it via Radicle. 174 199 // Nothing to do; next pass sees forgejoHead == radHead once 175 200 // it arrives, same as any other convergence. 201 + g.aiSkipStreak++ 176 202 break 177 203 } 178 204 if err := g.pushBranch("rad", "forgejo", forgejoHead, state.RadicleToForgejo); err != nil { ··· 188 214 189 215 case g.isAncestor(forgejoHead, radHead): 190 216 // Symmetric case: forgejo's tip is contained in rad's. 191 - if g.AuthorizedIntegrationSource != "" { 217 + if g.AuthorizedIntegrationSource != "" && g.aiSkipStreak < maxAISkipStreak { 218 + g.aiSkipStreak++ 192 219 break 193 220 } 194 221 if err := g.pushBranch("rad", "forgejo", forgejoHead, state.RadicleToForgejo); err != nil {
+4 -1
internal/sync/patches.go
··· 2 2 3 3 import ( 4 4 "bytes" 5 + "context" 5 6 "fmt" 6 7 "os" 7 8 "os/exec" ··· 32 33 } 33 34 34 35 func (s *PatchSyncer) git(args ...string) (string, error) { 35 - cmd := exec.Command("git", args...) 36 + ctx, cancel := context.WithTimeout(context.Background(), execTimeout) 37 + defer cancel() 38 + cmd := exec.CommandContext(ctx, "git", args...) 36 39 cmd.Dir = s.WorkDir 37 40 // Must start from the inherited environment, not a bare slice: git 38 41 // needs PATH to resolve the git-remote-rad helper for rad:// remotes.
+9
internal/sync/syncer.go
··· 152 152 } 153 153 } 154 154 155 + // HasAuthorizedIntegrationSource reports whether this pair's Forgejo side 156 + // already receives pushes directly from another pair's Forgejo via 157 + // Authorized Integrations (set by SetAuthorizedIntegrationSource at 158 + // startup) — used by the dashboard to explain why this side shows no 159 + // graft-logged events even though it holds the mirrored content. 160 + func (rs *RepoSyncer) HasAuthorizedIntegrationSource() bool { 161 + return rs.git != nil && rs.git.AuthorizedIntegrationSource != "" 162 + } 163 + 155 164 // Series is the dashboard row this pair's activity groups under: its 156 165 // configured series, or its own name if unset. 157 166 func (rs *RepoSyncer) Series() string {