| 1 | package routing |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/base64" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "net/http" |
| 9 | "path" |
| 10 | "strings" |
| 11 | |
| 12 | drclient "github.com/ipfs/boxo/routing/http/client" |
| 13 | "github.com/ipfs/boxo/routing/http/contentrouter" |
| 14 | "github.com/ipfs/go-datastore" |
| 15 | logging "github.com/ipfs/go-log/v2" |
| 16 | version "github.com/ipfs/kubo" |
| 17 | "github.com/ipfs/kubo/config" |
| 18 | dht "github.com/libp2p/go-libp2p-kad-dht" |
| 19 | "github.com/libp2p/go-libp2p-kad-dht/dual" |
| 20 | "github.com/libp2p/go-libp2p-kad-dht/fullrt" |
| 21 | record "github.com/libp2p/go-libp2p-record" |
| 22 | routinghelpers "github.com/libp2p/go-libp2p-routing-helpers" |
| 23 | ic "github.com/libp2p/go-libp2p/core/crypto" |
| 24 | host "github.com/libp2p/go-libp2p/core/host" |
| 25 | "github.com/libp2p/go-libp2p/core/peer" |
| 26 | "github.com/libp2p/go-libp2p/core/routing" |
| 27 | ma "github.com/multiformats/go-multiaddr" |
| 28 | "go.opencensus.io/stats/view" |
| 29 | "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" |
| 30 | ) |
| 31 | |
| 32 | var log = logging.Logger("routing/delegated") |
| 33 | |
| 34 | // Parse creates a composed router from the custom routing configuration. |
| 35 | // |
| 36 | // EXPERIMENTAL: Custom routing (Routing.Type=custom with Routing.Routers and |
| 37 | // Routing.Methods) is for research and testing only, not production use. |
| 38 | // The configuration format and behavior may change without notice between |
| 39 | // releases. HTTP-only configurations cannot reliably provide content. |
| 40 | // See docs/delegated-routing.md for limitations. |
| 41 | func Parse(routers config.Routers, methods config.Methods, extraDHT *ExtraDHTParams, extraHTTP *ExtraHTTPParams) (routing.Routing, error) { |
| 42 | if err := methods.Check(); err != nil { |
| 43 | return nil, err |
| 44 | } |
| 45 | |
| 46 | createdRouters := make(map[string]routing.Routing) |
| 47 | finalRouter := &Composer{} |
| 48 | |
| 49 | // Create all needed routers from method names |
| 50 | for mn, m := range methods { |
| 51 | router, err := parse(make(map[string]bool), createdRouters, m.RouterName, routers, extraDHT, extraHTTP) |
| 52 | if err != nil { |
| 53 | return nil, err |
| 54 | } |
| 55 | |
| 56 | switch mn { |
| 57 | case config.MethodNamePutIPNS: |
| 58 | finalRouter.PutValueRouter = router |
| 59 | case config.MethodNameGetIPNS: |
| 60 | finalRouter.GetValueRouter = router |
| 61 | case config.MethodNameFindPeers: |
| 62 | finalRouter.FindPeersRouter = router |
| 63 | case config.MethodNameFindProviders: |
| 64 | finalRouter.FindProvidersRouter = router |
| 65 | case config.MethodNameProvide: |
| 66 | finalRouter.ProvideRouter = router |
| 67 | } |
| 68 | |
| 69 | log.Info("using method ", mn, " with router ", m.RouterName) |
| 70 | } |
| 71 | |
| 72 | return finalRouter, nil |
| 73 | } |
| 74 | |
| 75 | func parse(visited map[string]bool, |
| 76 | createdRouters map[string]routing.Routing, |
| 77 | routerName string, |
| 78 | routersCfg config.Routers, |
| 79 | extraDHT *ExtraDHTParams, |
| 80 | extraHTTP *ExtraHTTPParams, |
| 81 | ) (routing.Routing, error) { |
| 82 | // check if we already created it |
| 83 | r, ok := createdRouters[routerName] |
| 84 | if ok { |
| 85 | return r, nil |
| 86 | } |
| 87 | |
| 88 | // check if we are in a dep loop |
| 89 | if visited[routerName] { |
| 90 | return nil, fmt.Errorf("dependency loop creating router with name %q", routerName) |
| 91 | } |
| 92 | |
| 93 | // set node as visited |
| 94 | visited[routerName] = true |
| 95 | |
| 96 | cfg, ok := routersCfg[routerName] |
| 97 | if !ok { |
| 98 | return nil, fmt.Errorf("config for router with name %q not found", routerName) |
| 99 | } |
| 100 | |
| 101 | var router routing.Routing |
| 102 | var err error |
| 103 | switch cfg.Type { |
| 104 | case config.RouterTypeHTTP: |
| 105 | router, err = httpRoutingFromConfig(cfg.Router, extraHTTP) |
| 106 | case config.RouterTypeDHT: |
| 107 | router, err = dhtRoutingFromConfig(cfg.Router, extraDHT) |
| 108 | case config.RouterTypeParallel: |
| 109 | crp := cfg.Parameters.(*config.ComposableRouterParams) |
| 110 | var pr []*routinghelpers.ParallelRouter |
| 111 | for _, cr := range crp.Routers { |
| 112 | ri, err := parse(visited, createdRouters, cr.RouterName, routersCfg, extraDHT, extraHTTP) |
| 113 | if err != nil { |
| 114 | return nil, err |
| 115 | } |
| 116 | |
| 117 | pr = append(pr, &routinghelpers.ParallelRouter{ |
| 118 | Router: ri, |
| 119 | IgnoreError: cr.IgnoreErrors, |
| 120 | DoNotWaitForSearchValue: true, |
| 121 | Timeout: cr.Timeout.Duration, |
| 122 | ExecuteAfter: cr.ExecuteAfter.WithDefault(0), |
| 123 | }) |
| 124 | |
| 125 | } |
| 126 | |
| 127 | router = routinghelpers.NewComposableParallel(pr) |
| 128 | case config.RouterTypeSequential: |
| 129 | crp := cfg.Parameters.(*config.ComposableRouterParams) |
| 130 | var sr []*routinghelpers.SequentialRouter |
| 131 | for _, cr := range crp.Routers { |
| 132 | ri, err := parse(visited, createdRouters, cr.RouterName, routersCfg, extraDHT, extraHTTP) |
| 133 | if err != nil { |
| 134 | return nil, err |
| 135 | } |
| 136 | |
| 137 | sr = append(sr, &routinghelpers.SequentialRouter{ |
| 138 | Router: ri, |
| 139 | IgnoreError: cr.IgnoreErrors, |
| 140 | Timeout: cr.Timeout.Duration, |
| 141 | }) |
| 142 | |
| 143 | } |
| 144 | |
| 145 | router = routinghelpers.NewComposableSequential(sr) |
| 146 | default: |
| 147 | return nil, fmt.Errorf("unknown router type %q", cfg.Type) |
| 148 | } |
| 149 | |
| 150 | if err != nil { |
| 151 | return nil, err |
| 152 | } |
| 153 | |
| 154 | createdRouters[routerName] = router |
| 155 | |
| 156 | log.Info("created router ", routerName, " with params ", cfg.Parameters) |
| 157 | |
| 158 | return router, nil |
| 159 | } |
| 160 | |
| 161 | type ExtraHTTPParams struct { |
| 162 | PeerID string |
| 163 | AddrFunc func() []ma.Multiaddr // dynamic address resolver for provider records |
| 164 | PrivKeyB64 string |
| 165 | HTTPRetrieval bool |
| 166 | } |
| 167 | |
| 168 | func ConstructHTTPRouter(endpoint string, peerID string, addrFunc func() []ma.Multiaddr, privKey string, httpRetrieval bool) (routing.Routing, error) { |
| 169 | return httpRoutingFromConfig( |
| 170 | config.Router{ |
| 171 | Type: "http", |
| 172 | Parameters: &config.HTTPRouterParams{ |
| 173 | Endpoint: endpoint, |
| 174 | }, |
| 175 | }, |
| 176 | &ExtraHTTPParams{ |
| 177 | PeerID: peerID, |
| 178 | AddrFunc: addrFunc, |
| 179 | PrivKeyB64: privKey, |
| 180 | HTTPRetrieval: httpRetrieval, |
| 181 | }, |
| 182 | ) |
| 183 | } |
| 184 | |
| 185 | func httpRoutingFromConfig(conf config.Router, extraHTTP *ExtraHTTPParams) (routing.Routing, error) { |
| 186 | params := conf.Parameters.(*config.HTTPRouterParams) |
| 187 | if params.Endpoint == "" { |
| 188 | return nil, NewParamNeededErr("Endpoint", conf.Type) |
| 189 | } |
| 190 | |
| 191 | params.FillDefaults() |
| 192 | |
| 193 | // Increase per-host connection pool since we are making lots of concurrent requests. |
| 194 | transport := http.DefaultTransport.(*http.Transport).Clone() |
| 195 | transport.MaxIdleConns = 500 |
| 196 | transport.MaxIdleConnsPerHost = 100 |
| 197 | |
| 198 | delegateHTTPClient := &http.Client{ |
| 199 | Transport: &drclient.ResponseBodyLimitedTransport{ |
| 200 | RoundTripper: otelhttp.NewTransport(transport, |
| 201 | otelhttp.WithSpanNameFormatter(func(operation string, req *http.Request) string { |
| 202 | if req.Method == http.MethodGet { |
| 203 | switch { |
| 204 | case strings.HasPrefix(req.URL.Path, "/routing/v1/providers"): |
| 205 | return "DelegatedHTTPClient.FindProviders" |
| 206 | case strings.HasPrefix(req.URL.Path, "/routing/v1/peers"): |
| 207 | return "DelegatedHTTPClient.FindPeers" |
| 208 | case strings.HasPrefix(req.URL.Path, "/routing/v1/ipns"): |
| 209 | return "DelegatedHTTPClient.GetIPNS" |
| 210 | } |
| 211 | } else if req.Method == http.MethodPut { |
| 212 | switch { |
| 213 | case strings.HasPrefix(req.URL.Path, "/routing/v1/ipns"): |
| 214 | return "DelegatedHTTPClient.PutIPNS" |
| 215 | } |
| 216 | } |
| 217 | return "DelegatedHTTPClient." + path.Dir(req.URL.Path) |
| 218 | }), |
| 219 | ), |
| 220 | LimitBytes: 1 << 20, |
| 221 | }, |
| 222 | } |
| 223 | |
| 224 | key, err := decodePrivKey(extraHTTP.PrivKeyB64) |
| 225 | if err != nil { |
| 226 | return nil, err |
| 227 | } |
| 228 | |
| 229 | protocols := config.DefaultHTTPRoutersFilterProtocols |
| 230 | if extraHTTP.HTTPRetrieval { |
| 231 | protocols = append(protocols, "transport-ipfs-gateway-http") |
| 232 | } |
| 233 | |
| 234 | peerID, err := peer.Decode(extraHTTP.PeerID) |
| 235 | if err != nil { |
| 236 | return nil, err |
| 237 | } |
| 238 | |
| 239 | var providerInfoOpt drclient.Option |
| 240 | if extraHTTP.AddrFunc != nil { |
| 241 | providerInfoOpt = drclient.WithProviderInfoFunc(peerID, extraHTTP.AddrFunc) |
| 242 | } else { |
| 243 | providerInfoOpt = drclient.WithProviderInfo(peerID, nil) |
| 244 | } |
| 245 | |
| 246 | cli, err := drclient.New( |
| 247 | params.Endpoint, |
| 248 | drclient.WithHTTPClient(delegateHTTPClient), |
| 249 | drclient.WithIdentity(key), |
| 250 | providerInfoOpt, |
| 251 | drclient.WithUserAgent(version.GetUserAgentVersion()), |
| 252 | drclient.WithProtocolFilter(protocols), |
| 253 | drclient.WithStreamResultsRequired(), // https://specs.ipfs.tech/routing/http-routing-v1/#streaming |
| 254 | drclient.WithDisabledLocalFiltering(false), // force local filtering in case remote server does not support IPIP-484 |
| 255 | ) |
| 256 | if err != nil { |
| 257 | return nil, err |
| 258 | } |
| 259 | |
| 260 | cr := contentrouter.NewContentRoutingClient( |
| 261 | cli, |
| 262 | contentrouter.WithMaxProvideBatchSize(params.MaxProvideBatchSize), |
| 263 | contentrouter.WithMaxProvideConcurrency(params.MaxProvideConcurrency), |
| 264 | ) |
| 265 | |
| 266 | err = view.Register(drclient.OpenCensusViews...) |
| 267 | if err != nil { |
| 268 | return nil, fmt.Errorf("registering HTTP delegated routing views: %w", err) |
| 269 | } |
| 270 | |
| 271 | return &httpRoutingWrapper{ |
| 272 | ContentRouting: cr, |
| 273 | PeerRouting: cr, |
| 274 | ValueStore: cr, |
| 275 | ProvideManyRouter: cr, |
| 276 | }, nil |
| 277 | } |
| 278 | |
| 279 | func decodePrivKey(keyB64 string) (ic.PrivKey, error) { |
| 280 | pk, err := base64.StdEncoding.DecodeString(keyB64) |
| 281 | if err != nil { |
| 282 | return nil, err |
| 283 | } |
| 284 | |
| 285 | return ic.UnmarshalPrivateKey(pk) |
| 286 | } |
| 287 | |
| 288 | type ExtraDHTParams struct { |
| 289 | BootstrapPeers []peer.AddrInfo |
| 290 | Host host.Host |
| 291 | Validator record.Validator |
| 292 | Datastore datastore.Batching |
| 293 | Context context.Context |
| 294 | } |
| 295 | |
| 296 | func dhtRoutingFromConfig(conf config.Router, extra *ExtraDHTParams) (routing.Routing, error) { |
| 297 | params, ok := conf.Parameters.(*config.DHTRouterParams) |
| 298 | if !ok { |
| 299 | return nil, errors.New("incorrect params for DHT router") |
| 300 | } |
| 301 | |
| 302 | if params.AcceleratedDHTClient { |
| 303 | return createFullRT(extra) |
| 304 | } |
| 305 | |
| 306 | var mode dht.ModeOpt |
| 307 | switch params.Mode { |
| 308 | case config.DHTModeAuto: |
| 309 | mode = dht.ModeAuto |
| 310 | case config.DHTModeClient: |
| 311 | mode = dht.ModeClient |
| 312 | case config.DHTModeServer: |
| 313 | mode = dht.ModeServer |
| 314 | default: |
| 315 | return nil, fmt.Errorf("invalid DHT mode: %q", params.Mode) |
| 316 | } |
| 317 | |
| 318 | return createDHT(extra, params.PublicIPNetwork, mode) |
| 319 | } |
| 320 | |
| 321 | func createDHT(params *ExtraDHTParams, public bool, mode dht.ModeOpt) (routing.Routing, error) { |
| 322 | var opts []dht.Option |
| 323 | |
| 324 | if public { |
| 325 | opts = append(opts, dht.QueryFilter(dht.PublicQueryFilter), |
| 326 | dht.RoutingTableFilter(dht.PublicRoutingTableFilter), |
| 327 | dht.RoutingTablePeerDiversityFilter(dht.NewRTPeerDiversityFilter(params.Host, 2, 3))) |
| 328 | } else { |
| 329 | opts = append(opts, dht.ProtocolExtension(dual.LanExtension), |
| 330 | dht.QueryFilter(dht.PrivateQueryFilter), |
| 331 | dht.RoutingTableFilter(dht.PrivateRoutingTableFilter)) |
| 332 | } |
| 333 | |
| 334 | opts = append(opts, |
| 335 | dht.Concurrency(10), |
| 336 | dht.Mode(mode), |
| 337 | dht.Datastore(params.Datastore), |
| 338 | dht.Validator(params.Validator), |
| 339 | dht.BootstrapPeers(params.BootstrapPeers...)) |
| 340 | |
| 341 | return dht.New( |
| 342 | params.Context, params.Host, opts..., |
| 343 | ) |
| 344 | } |
| 345 | |
| 346 | func createFullRT(params *ExtraDHTParams) (routing.Routing, error) { |
| 347 | return fullrt.NewFullRT(params.Host, |
| 348 | dht.DefaultPrefix, |
| 349 | fullrt.DHTOption( |
| 350 | dht.Validator(params.Validator), |
| 351 | dht.Datastore(params.Datastore), |
| 352 | dht.BootstrapPeers(params.BootstrapPeers...), |
| 353 | dht.BucketSize(20), |
| 354 | ), |
| 355 | ) |
| 356 | } |