Monorepo for Tangled
tangled.org
1package xrpc
2
3import (
4 "cmp"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "maps"
10 "net/http"
11 "net/url"
12 "path"
13 "strings"
14
15 "github.com/bluesky-social/indigo/atproto/atclient"
16 "github.com/bluesky-social/indigo/atproto/syntax"
17 indigoxrpc "github.com/bluesky-social/indigo/xrpc"
18 "github.com/go-git/go-git/v5/plumbing/filemode"
19 "tangled.org/core/api/tangled"
20 "tangled.org/core/knotmirror/db"
21 "tangled.org/core/knotmirror/models"
22)
23
24var mirrorToKnotNSID = map[string]string{
25 tangled.GitTempListBranchesNSID: tangled.RepoBranchesNSID,
26 tangled.GitTempListTagsNSID: tangled.RepoTagsNSID,
27 tangled.GitTempListCommitsNSID: tangled.RepoLogNSID,
28 tangled.GitTempGetTreeNSID: tangled.RepoTreeNSID,
29 tangled.GitTempGetBranchNSID: tangled.RepoBranchNSID,
30 tangled.GitTempGetTagNSID: tangled.RepoTagNSID,
31 tangled.GitTempGetArchiveNSID: tangled.RepoArchiveNSID,
32 tangled.GitTempListLanguagesNSID: tangled.RepoLanguagesNSID,
33 tangled.GitTempGetBlobNSID: tangled.RepoBlobNSID,
34}
35
36var hopByHopHeaders = map[string]bool{
37 "Connection": true,
38 "Keep-Alive": true,
39 "Transfer-Encoding": true,
40 "Te": true,
41 "Trailer": true,
42 "Upgrade": true,
43 "Proxy-Authorization": true,
44 "Proxy-Authenticate": true,
45}
46
47type knotInfo struct {
48 baseURL string
49 repoIdentifier string
50}
51
52// validateKnotURL ensures a knot base URL is safe to proxy to.
53// It rejects URLs with path components, query strings, or fragments
54// that could be used for path injection.
55func validateKnotURL(raw string) (string, error) {
56 u, err := url.Parse(raw)
57 if err != nil {
58 return "", fmt.Errorf("invalid knot URL: %w", err)
59 }
60 if u.Scheme != "http" && u.Scheme != "https" {
61 return "", errors.New("knot URL must use http or https scheme")
62 }
63 if u.Path != "" && u.Path != "/" {
64 return "", fmt.Errorf("knot URL must not contain a path: %q", raw)
65 }
66 if u.RawQuery != "" || u.Fragment != "" {
67 return "", fmt.Errorf("knot URL must not contain query or fragment: %q", raw)
68 }
69 if u.User != nil {
70 return "", fmt.Errorf("knot URL must not contain userinfo: %q", raw)
71 }
72 // Strip trailing slash for consistent formatting
73 return strings.TrimRight(u.String(), "/"), nil
74}
75
76func (x *Xrpc) resolveKnot(ctx context.Context, repoDid syntax.DID) (*knotInfo, error) {
77 if repo, err := db.GetRepoByRepoDid(ctx, x.db, repoDid); err == nil && repo != nil {
78 knotURL := repo.KnotDomain
79 if !strings.Contains(repo.KnotDomain, "://") {
80 if host, _ := db.GetHost(ctx, x.db, repo.KnotDomain); host != nil {
81 knotURL = host.URL()
82 } else {
83 x.logger.Warn("repo is from unknown knot")
84 if x.cfg.KnotUseSSL {
85 knotURL = "https://" + knotURL
86 } else {
87 knotURL = "http://" + knotURL
88 }
89 }
90 }
91 knotURL, err = validateKnotURL(knotURL)
92 if err != nil {
93 return nil, err
94 }
95 return &knotInfo{baseURL: knotURL, repoIdentifier: repo.RepoIdentifier()}, nil
96 }
97
98 ident, err := x.resolver.ResolveIdent(ctx, repoDid.String())
99 if err != nil {
100 return nil, fmt.Errorf("resolving repoDid %s: %w", repoDid, err)
101 }
102 knotURL, err := validateKnotURL(ident.GetServiceEndpoint("atproto_pds"))
103 if err != nil {
104 return nil, fmt.Errorf("repoDid %s: %w", repoDid, err)
105 }
106
107 xrpcc := &indigoxrpc.Client{Host: knotURL, Client: x.httpClient}
108 out, err := tangled.RepoDescribeRepo(ctx, xrpcc, repoDid.String())
109 if err != nil {
110 x.logger.Warn("describeRepo failed; serving without metadata upsert", "knot", knotURL, "repo", repoDid, "err", err)
111 return &knotInfo{baseURL: knotURL, repoIdentifier: repoDid.String()}, nil
112 }
113 if out.RepoDid != repoDid.String() {
114 return nil, fmt.Errorf("knot %s returned mismatched repoDid: got %q, want %q", knotURL, out.RepoDid, repoDid)
115 }
116 ownerDid, err := syntax.ParseDID(out.OwnerDid)
117 if err != nil {
118 return nil, fmt.Errorf("describeRepo on %s returned invalid ownerDid %q: %w", knotURL, out.OwnerDid, err)
119 }
120 rkey, err := syntax.ParseRecordKey(out.Rkey)
121 if err != nil {
122 return nil, fmt.Errorf("describeRepo on %s returned invalid rkey %q: %w", knotURL, out.Rkey, err)
123 }
124
125 go func() {
126 pending := &models.Repo{
127 Did: ownerDid,
128 Rkey: rkey,
129 Name: string(rkey),
130 KnotDomain: knotURL,
131 RepoDid: repoDid,
132 State: models.RepoStatePending,
133 }
134 if err := db.UpsertRepo(context.Background(), x.db, pending); err != nil {
135 x.logger.Error("failed to upsert repo after directory resolution", "err", err)
136 }
137 }()
138
139 return &knotInfo{baseURL: knotURL, repoIdentifier: repoDid.String()}, nil
140}
141
142func (x *Xrpc) proxyToKnot(w http.ResponseWriter, r *http.Request, repoDid syntax.DID) bool {
143 mirrorNSID := strings.TrimPrefix(r.URL.Path, "/xrpc/")
144 knotNSID, ok := mirrorToKnotNSID[mirrorNSID]
145 if !ok {
146 return false
147 }
148
149 knot, err := x.resolveKnot(r.Context(), repoDid)
150 if err != nil {
151 x.logger.Warn("proxy: failed to resolve knot", "repo", repoDid, "err", err)
152 return false
153 }
154
155 params := make(url.Values)
156 maps.Copy(params, r.URL.Query())
157 params.Set("repo", knot.repoIdentifier)
158
159 target := fmt.Sprintf("%s/xrpc/%s?%s", knot.baseURL, knotNSID, params.Encode())
160
161 req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, target, nil)
162 if err != nil {
163 x.logger.Warn("proxy: failed to build request", "target", target, "err", err)
164 return false
165 }
166
167 resp, err := x.httpClient.Do(req)
168 if err != nil {
169 x.logger.Warn("proxy: knot request failed", "target", target, "err", err)
170 return false
171 }
172 defer resp.Body.Close()
173
174 for k, vv := range resp.Header {
175 if hopByHopHeaders[k] {
176 continue
177 }
178 for _, v := range vv {
179 w.Header().Add(k, v)
180 }
181 }
182 w.WriteHeader(resp.StatusCode)
183 if _, err := io.Copy(w, resp.Body); err != nil {
184 x.logger.Warn("proxy: response copy interrupted", "target", target, "err", err)
185 }
186
187 x.logger.Info("proxy: served from knot", "repo", repoDid, "knot", knot.baseURL, "status", resp.StatusCode)
188 return true
189}
190
191func (x *Xrpc) forwardSuspended(next http.Handler) http.Handler {
192 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
193 repoDid, err := syntax.ParseDID(r.URL.Query().Get("repo"))
194 if err != nil {
195 next.ServeHTTP(w, r)
196 return
197 }
198
199 repo, err := db.GetRepoByRepoDid(r.Context(), x.db, repoDid)
200 if err != nil || repo == nil || repo.State != models.RepoStateSuspended {
201 next.ServeHTTP(w, r)
202 return
203 }
204
205 nsid := strings.TrimPrefix(r.URL.Path, "/xrpc/")
206 switch nsid {
207 case tangled.GitTempGetEntryNSID:
208 x.serveSuspendedEntry(w, r, repoDid)
209 case tangled.GitTempGetBlobNSID:
210 q := r.URL.Query()
211 q.Set("raw", "true")
212 r.URL.RawQuery = q.Encode()
213 x.forwardOrFail(w, r, repoDid)
214 default:
215 if _, ok := mirrorToKnotNSID[nsid]; !ok {
216 next.ServeHTTP(w, r)
217 return
218 }
219 x.forwardOrFail(w, r, repoDid)
220 }
221 })
222}
223
224func (x *Xrpc) forwardOrFail(w http.ResponseWriter, r *http.Request, repoDid syntax.DID) {
225 if x.proxyToKnot(w, r, repoDid) {
226 return
227 }
228 writeJson(w, http.StatusBadGateway, atclient.ErrorBody{Name: "BadGateway", Message: "failed to reach knot for suspended repo"})
229}
230
231func (x *Xrpc) serveSuspendedEntry(w http.ResponseWriter, r *http.Request, repoDid syntax.DID) {
232 ref := cmp.Or(r.URL.Query().Get("ref"), "HEAD")
233 filePath := r.URL.Query().Get("path")
234 if filePath == "" {
235 writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "missing path parameter"})
236 return
237 }
238
239 knot, err := x.resolveKnot(r.Context(), repoDid)
240 if err != nil {
241 x.logger.Warn("suspended entry: failed to resolve knot", "repo", repoDid, "err", err)
242 writeJson(w, http.StatusBadGateway, atclient.ErrorBody{Name: "BadGateway", Message: "failed to resolve knot for suspended repo"})
243 return
244 }
245
246 client := &indigoxrpc.Client{Host: knot.baseURL, Client: x.httpClient}
247 out, err := tangled.RepoBlob(r.Context(), client, filePath, false, ref, knot.repoIdentifier)
248 if err != nil {
249 x.logger.Warn("suspended entry: knot repo.blob failed", "repo", repoDid, "err", err)
250 writeJson(w, http.StatusBadGateway, atclient.ErrorBody{Name: "BadGateway", Message: "failed to read entry from knot"})
251 return
252 }
253
254 mode := filemode.Regular
255 if out.Submodule != nil {
256 mode = filemode.Submodule
257 }
258
259 writeJson(w, http.StatusOK, tangled.GitTempGetEntry_Output{
260 Name: path.Base(filePath),
261 Mode: mode.String(),
262 Size: derefInt64(out.Size),
263 LastCommit: suspendedLastCommit(out.LastCommit),
264 Submodule: suspendedSubmodule(out.Submodule),
265 })
266}
267
268func suspendedLastCommit(c *tangled.RepoBlob_LastCommit) *tangled.GitTempDefs_Commit {
269 if c == nil || c.Author == nil {
270 return nil
271 }
272 sig := suspendedSignature(c.Author)
273 hash := c.Hash
274 return &tangled.GitTempDefs_Commit{
275 Author: sig,
276 Committer: sig,
277 Hash: &hash,
278 Message: c.Message,
279 }
280}
281
282func suspendedSignature(s *tangled.RepoBlob_Signature) *tangled.GitTempDefs_Signature {
283 if s == nil {
284 return nil
285 }
286 return &tangled.GitTempDefs_Signature{
287 Name: s.Name,
288 Email: s.Email,
289 When: s.When,
290 }
291}
292
293func suspendedSubmodule(s *tangled.RepoBlob_Submodule) *tangled.GitTempDefs_Submodule {
294 if s == nil {
295 return nil
296 }
297 return &tangled.GitTempDefs_Submodule{
298 Name: s.Name,
299 Url: s.Url,
300 Branch: s.Branch,
301 }
302}
303
304func derefInt64(v *int64) int64 {
305 if v == nil {
306 return 0
307 }
308 return *v
309}