master
go 323 lines 9.42 KB
Raw
1 package migrations
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "net"
9 "net/http"
10 gopath "path"
11 "strings"
12 "time"
13
14 "github.com/ipfs/boxo/blockservice"
15 "github.com/ipfs/boxo/blockstore"
16 "github.com/ipfs/boxo/exchange/offline"
17 bsfetcher "github.com/ipfs/boxo/fetcher/impl/blockservice"
18 files "github.com/ipfs/boxo/files"
19 "github.com/ipfs/boxo/ipld/merkledag"
20 unixfile "github.com/ipfs/boxo/ipld/unixfs/file"
21 "github.com/ipfs/boxo/ipns"
22 "github.com/ipfs/boxo/namesys"
23 "github.com/ipfs/boxo/path"
24 "github.com/ipfs/boxo/path/resolver"
25 "github.com/ipfs/go-datastore"
26 dssync "github.com/ipfs/go-datastore/sync"
27 "github.com/ipfs/go-unixfsnode"
28 config "github.com/ipfs/kubo/config"
29 gocarv2 "github.com/ipld/go-car/v2"
30 dagpb "github.com/ipld/go-codec-dagpb"
31 madns "github.com/multiformats/go-multiaddr-dns"
32 )
33
34 const (
35 // defaultGatewayURL is used when no gateway is configured. Some ISPs
36 // block ipfs.io, so we use a different hostname.
37 defaultGatewayURL = "https://trustless-gateway.link"
38 // Default maximum download size.
39 defaultFetchLimit = 1024 * 1024 * 512
40
41 // Sized for users on slow / high-latency networks (e.g. VPNs through
42 // congested links): a 3-RTT TLS handshake at 1-2s RTT plus packet
43 // loss can legitimately approach 10s. 15s leaves headroom while
44 // still failing fast against truly dead gateways.
45 dialTimeout = 15 * time.Second
46 tlsHandshakeTimeout = 15 * time.Second
47 )
48
49 // defaultMigrationGateways is a last-resort fallback used when
50 // Migration.DownloadSources expands the "HTTPS" alias. The first entry
51 // (trustless-gateway.link) serves nearly all users; the rest are tried
52 // only when it is blocked or unreachable.
53 //
54 // Including third-party gateways is safe: each block is fetched as CAR
55 // and verified against the requested CID's multihash, so a malicious
56 // operator cannot substitute different content.
57 //
58 // TODO: replace this static list with a dynamic source, either the public
59 // gateway checker list at
60 // https://github.com/ipfs/public-gateway-checker/raw/refs/heads/main/gateways.json
61 // or AutoConf. Not done yet because this code path only runs for repos
62 // from go-ipfs or Kubo older than v0.27 (roughly 2020 vintage). Modern
63 // Kubo ships embedded migrations and never reaches it, so the impact and
64 // risk of leaving the list hard-coded are both low.
65 var defaultMigrationGateways = []string{
66 defaultGatewayURL,
67 "https://gateway.pinata.cloud",
68 "https://ipfs.filebase.io",
69 "https://4everland.io",
70 "https://dget.top",
71 }
72
73 // migrationHTTPClient is the HTTP client for migration downloads. Its timeouts
74 // fail fast on unreachable or stalled gateways so MultiFetcher can rotate.
75 var migrationHTTPClient = &http.Client{
76 Transport: &http.Transport{
77 Proxy: http.ProxyFromEnvironment,
78 DialContext: (&net.Dialer{
79 Timeout: dialTimeout,
80 }).DialContext,
81 TLSHandshakeTimeout: tlsHandshakeTimeout,
82 // ResponseHeaderTimeout matches boxo's server-side
83 // DefaultRetrievalTimeout (re-exported via Kubo config): a healthy
84 // gateway returns the first byte within this budget or 504s.
85 // Mirroring it avoids drift if boxo retunes.
86 ResponseHeaderTimeout: config.DefaultRetrievalTimeout,
87 IdleConnTimeout: 90 * time.Second,
88 ExpectContinueTimeout: 1 * time.Second,
89 ForceAttemptHTTP2: true,
90 },
91 // No overall Timeout: cancellation flows from the request context, so
92 // streaming bodies run to completion under context control.
93 }
94
95 // HttpFetcher fetches files over HTTP using verifiable CAR archives.
96 type HttpFetcher struct { //nolint
97 distPath string
98 gateway string
99 limit int64
100 userAgent string
101 }
102
103 var _ Fetcher = (*HttpFetcher)(nil)
104
105 // NewHttpFetcher creates a new [HttpFetcher].
106 //
107 // Specifying "" for distPath sets the default IPNS path.
108 // Specifying "" for gateway sets the default.
109 // Specifying 0 for fetchLimit sets the default, -1 means no limit.
110 func NewHttpFetcher(distPath, gateway, userAgent string, fetchLimit int64) *HttpFetcher { //nolint
111 f := &HttpFetcher{
112 distPath: LatestIpfsDist,
113 gateway: defaultGatewayURL,
114 limit: defaultFetchLimit,
115 userAgent: userAgent,
116 }
117
118 if distPath != "" {
119 if !strings.HasPrefix(distPath, "/") {
120 distPath = "/" + distPath
121 }
122 f.distPath = distPath
123 }
124
125 if gateway != "" {
126 f.gateway = strings.TrimRight(gateway, "/")
127 }
128
129 if fetchLimit != 0 {
130 if fetchLimit < 0 {
131 fetchLimit = 0
132 }
133 f.limit = fetchLimit
134 }
135
136 return f
137 }
138
139 // Fetch attempts to fetch the file at the given path, from the distribution
140 // site configured for this HttpFetcher.
141 func (f *HttpFetcher) Fetch(ctx context.Context, filePath string) ([]byte, error) {
142 imPath, err := f.resolvePath(ctx, gopath.Join(f.distPath, filePath))
143 if err != nil {
144 return nil, fmt.Errorf("path could not be resolved: %w", err)
145 }
146
147 rc, err := f.httpRequest(ctx, imPath, "application/vnd.ipld.car", "car")
148 if err != nil {
149 return nil, fmt.Errorf("failed to fetch CAR: %w", err)
150 }
151
152 return carStreamToFileBytes(ctx, rc, imPath)
153 }
154
155 func (f *HttpFetcher) Close() error {
156 return nil
157 }
158
159 func (f *HttpFetcher) resolvePath(ctx context.Context, pathStr string) (path.ImmutablePath, error) {
160 p, err := path.NewPath(pathStr)
161 if err != nil {
162 return path.ImmutablePath{}, fmt.Errorf("path is invalid: %w", err)
163 }
164
165 for p.Mutable() {
166 // Download IPNS record and verify through the gateway, or resolve the
167 // DNSLink with the default DNS resolver.
168 name, err := ipns.NameFromString(p.Segments()[1])
169 if err == nil {
170 p, err = f.resolveIPNS(ctx, name)
171 } else {
172 p, err = f.resolveDNSLink(ctx, p)
173 }
174
175 if err != nil {
176 return path.ImmutablePath{}, err
177 }
178 }
179
180 return path.NewImmutablePath(p)
181 }
182
183 func (f *HttpFetcher) resolveIPNS(ctx context.Context, name ipns.Name) (path.Path, error) {
184 rc, err := f.httpRequest(ctx, name.AsPath(), "application/vnd.ipfs.ipns-record", "ipns-record")
185 if err != nil {
186 return path.ImmutablePath{}, err
187 }
188 defer rc.Close()
189
190 rc = NewLimitReadCloser(rc, int64(ipns.MaxRecordSize))
191 rawRecord, err := io.ReadAll(rc)
192 if err != nil {
193 return path.ImmutablePath{}, err
194 }
195
196 rec, err := ipns.UnmarshalRecord(rawRecord)
197 if err != nil {
198 return path.ImmutablePath{}, err
199 }
200
201 err = ipns.ValidateWithName(rec, name)
202 if err != nil {
203 return path.ImmutablePath{}, err
204 }
205
206 return rec.Value()
207 }
208
209 func (f *HttpFetcher) resolveDNSLink(ctx context.Context, p path.Path) (path.Path, error) {
210 dnsResolver := namesys.NewDNSResolver(madns.DefaultResolver.LookupTXT)
211 res, err := dnsResolver.Resolve(ctx, p)
212 if err != nil {
213 return nil, err
214 }
215 return res.Path, nil
216 }
217
218 func (f *HttpFetcher) httpRequest(ctx context.Context, p path.Path, accept, format string) (io.ReadCloser, error) {
219 url := f.gateway + p.String()
220 // Pass the format hint as both an Accept header and a ?format= query
221 // parameter. The trustless gateway spec defines both as valid
222 // signaling mechanisms, and some gateway implementations honor one
223 // but not the other; sending both maximizes compatibility.
224 if format != "" {
225 url += "?format=" + format
226 }
227 fmt.Printf("Fetching with HTTP: %q\n", url)
228 req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
229 if err != nil {
230 return nil, fmt.Errorf("http.NewRequest error: %w", err)
231 }
232 req.Header.Set("Accept", accept)
233
234 if f.userAgent != "" {
235 req.Header.Set("User-Agent", f.userAgent)
236 }
237
238 resp, err := migrationHTTPClient.Do(req)
239 if err != nil {
240 return nil, fmt.Errorf("migration http request error: %w", err)
241 }
242
243 if resp.StatusCode >= 400 {
244 defer resp.Body.Close()
245 mes, err := io.ReadAll(resp.Body)
246 if err != nil {
247 return nil, fmt.Errorf("error reading error body: %w", err)
248 }
249 return nil, fmt.Errorf("GET %s error: %s: %s", url, resp.Status, string(mes))
250 }
251
252 var rc io.ReadCloser
253 if f.limit != 0 {
254 rc = NewLimitReadCloser(resp.Body, f.limit)
255 } else {
256 rc = resp.Body
257 }
258
259 return rc, nil
260 }
261
262 func carStreamToFileBytes(ctx context.Context, r io.ReadCloser, imPath path.ImmutablePath) ([]byte, error) {
263 defer r.Close()
264
265 // Create temporary block datastore and dag service.
266 dataStore := dssync.MutexWrap(datastore.NewMapDatastore())
267 blockStore := blockstore.NewBlockstore(dataStore)
268 blockService := blockservice.New(blockStore, offline.Exchange(blockStore))
269 dagService := merkledag.NewDAGService(blockService)
270
271 defer dagService.Blocks.Close()
272 defer dataStore.Close()
273
274 // Create CAR reader
275 car, err := gocarv2.NewBlockReader(r)
276 if err != nil {
277 fmt.Println(err)
278 return nil, fmt.Errorf("error creating car reader: %s", err)
279 }
280
281 // Add all blocks to the blockstore.
282 for {
283 block, err := car.Next()
284 if err != nil && err != io.EOF {
285 return nil, fmt.Errorf("error reading block from car: %s", err)
286 } else if block == nil {
287 break
288 }
289
290 err = blockStore.Put(ctx, block)
291 if err != nil {
292 return nil, fmt.Errorf("error putting block in blockstore: %s", err)
293 }
294 }
295
296 fetcherCfg := bsfetcher.NewFetcherConfig(blockService)
297 fetcherCfg.PrototypeChooser = dagpb.AddSupportToChooser(bsfetcher.DefaultPrototypeChooser)
298 fetcher := fetcherCfg.WithReifier(unixfsnode.Reify)
299 resolver := resolver.NewBasicResolver(fetcher)
300
301 cid, _, err := resolver.ResolveToLastNode(ctx, imPath)
302 if err != nil {
303 return nil, fmt.Errorf("failed to resolve: %w", err)
304 }
305
306 nd, err := dagService.Get(ctx, cid)
307 if err != nil {
308 return nil, fmt.Errorf("failed to resolve: %w", err)
309 }
310
311 // Make UnixFS file out of the node.
312 uf, err := unixfile.NewUnixfsFile(ctx, dagService, nd)
313 if err != nil {
314 return nil, fmt.Errorf("error building unixfs file: %s", err)
315 }
316
317 // Check if it's a file and return.
318 if f, ok := uf.(files.File); ok {
319 return io.ReadAll(f)
320 }
321
322 return nil, errors.New("unexpected unixfs node type")
323 }