@cryptotaxi247 / kubo
Code Commits Issues Pulls Sessions CI/CD
master
master
go 194 lines 4.91 KB
Raw
1 package corehttp
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "net"
8 "net/http"
9 "time"
10
11 "github.com/ipfs/boxo/gateway"
12 "github.com/ipfs/boxo/ipns"
13 "github.com/ipfs/boxo/routing/http/server"
14 "github.com/ipfs/boxo/routing/http/types"
15 "github.com/ipfs/boxo/routing/http/types/iter"
16 cid "github.com/ipfs/go-cid"
17 core "github.com/ipfs/kubo/core"
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 "github.com/libp2p/go-libp2p/core/peer"
22 "github.com/libp2p/go-libp2p/core/routing"
23 )
24
25 func RoutingOption() ServeOption {
26 return func(n *core.IpfsNode, _ net.Listener, mux *http.ServeMux) (*http.ServeMux, error) {
27 _, headers, err := getGatewayConfig(n)
28 if err != nil {
29 return nil, err
30 }
31
32 handler := server.Handler(&contentRouter{n})
33 handler = gateway.NewHeaders(headers).ApplyCors().Wrap(handler)
34 mux.Handle("/routing/v1/", handler)
35 return mux, nil
36 }
37 }
38
39 type contentRouter struct {
40 n *core.IpfsNode
41 }
42
43 func (r *contentRouter) FindProviders(ctx context.Context, key cid.Cid, limit int) (iter.ResultIter[types.Record], error) {
44 ctx, cancel := context.WithCancel(ctx)
45 ch := r.n.Routing.FindProvidersAsync(ctx, key, limit)
46 return iter.ToResultIter[types.Record](&peerChanIter{
47 ch: ch,
48 cancel: cancel,
49 }), nil
50 }
51
52 // nolint deprecated
53 func (r *contentRouter) ProvideBitswap(ctx context.Context, req *server.BitswapWriteProvideRequest) (time.Duration, error) {
54 return 0, routing.ErrNotSupported
55 }
56
57 func (r *contentRouter) FindPeers(ctx context.Context, pid peer.ID, limit int) (iter.ResultIter[*types.PeerRecord], error) {
58 ctx, cancel := context.WithCancel(ctx)
59 defer cancel()
60
61 addr, err := r.n.Routing.FindPeer(ctx, pid)
62 if err != nil {
63 return nil, err
64 }
65
66 rec := &types.PeerRecord{
67 Schema: types.SchemaPeer,
68 ID: &addr.ID,
69 }
70
71 for _, addr := range addr.Addrs {
72 rec.Addrs = append(rec.Addrs, types.Multiaddr{Multiaddr: addr})
73 }
74
75 return iter.ToResultIter[*types.PeerRecord](iter.FromSlice[*types.PeerRecord]([]*types.PeerRecord{rec})), nil
76 }
77
78 func (r *contentRouter) GetIPNS(ctx context.Context, name ipns.Name) (*ipns.Record, error) {
79 ctx, cancel := context.WithCancel(ctx)
80 defer cancel()
81
82 raw, err := r.n.Routing.GetValue(ctx, string(name.RoutingKey()))
83 if err != nil {
84 return nil, err
85 }
86
87 return ipns.UnmarshalRecord(raw)
88 }
89
90 func (r *contentRouter) PutIPNS(ctx context.Context, name ipns.Name, record *ipns.Record) error {
91 ctx, cancel := context.WithCancel(ctx)
92 defer cancel()
93
94 raw, err := ipns.MarshalRecord(record)
95 if err != nil {
96 return err
97 }
98
99 // The caller guarantees that name matches the record. This is double checked
100 // by the internals of PutValue.
101 return r.n.Routing.PutValue(ctx, string(name.RoutingKey()), raw)
102 }
103
104 func (r *contentRouter) GetClosestPeers(ctx context.Context, key cid.Cid) (iter.ResultIter[*types.PeerRecord], error) {
105 // Per the spec, if the peer ID is empty, we should use self.
106 if key == cid.Undef {
107 return nil, errors.New("GetClosestPeers key is undefined")
108 }
109
110 keyStr := string(key.Hash())
111 var peers []peer.ID
112 var err error
113
114 if r.n.DHTClient == nil {
115 return nil, fmt.Errorf("GetClosestPeers not supported: DHT is not available")
116 }
117
118 switch dhtClient := r.n.DHTClient.(type) {
119 case *dual.DHT:
120 // Only use WAN DHT for public HTTP Routing API.
121 // LAN DHT contains private network peers that should not be exposed publicly.
122 if dhtClient.WAN == nil {
123 return nil, fmt.Errorf("GetClosestPeers not supported: WAN DHT is not available")
124 }
125 peers, err = dhtClient.WAN.GetClosestPeers(ctx, keyStr)
126 case *fullrt.FullRT:
127 peers, err = dhtClient.GetClosestPeers(ctx, keyStr)
128 case *dht.IpfsDHT:
129 peers, err = dhtClient.GetClosestPeers(ctx, keyStr)
130 default:
131 return nil, fmt.Errorf("GetClosestPeers not supported for DHT type %T", r.n.DHTClient)
132 }
133
134 if err != nil {
135 return nil, err
136 }
137
138 // We have some DHT-closest peers. Find addresses for them.
139 // The addresses should be in the peerstore.
140 records := make([]*types.PeerRecord, 0, len(peers))
141 for _, p := range peers {
142 addrs := r.n.Peerstore.Addrs(p)
143 rAddrs := make([]types.Multiaddr, len(addrs))
144 for i, addr := range addrs {
145 rAddrs[i] = types.Multiaddr{Multiaddr: addr}
146 }
147 record := types.PeerRecord{
148 ID: &p,
149 Schema: types.SchemaPeer,
150 Addrs: rAddrs,
151 }
152 records = append(records, &record)
153 }
154
155 return iter.ToResultIter(iter.FromSlice(records)), nil
156 }
157
158 type peerChanIter struct {
159 ch <-chan peer.AddrInfo
160 cancel context.CancelFunc
161 next *peer.AddrInfo
162 }
163
164 func (it *peerChanIter) Next() bool {
165 addr, ok := <-it.ch
166 if ok {
167 it.next = &addr
168 return true
169 }
170 it.next = nil
171 return false
172 }
173
174 func (it *peerChanIter) Val() types.Record {
175 if it.next == nil {
176 return nil
177 }
178
179 rec := &types.PeerRecord{
180 Schema: types.SchemaPeer,
181 ID: &it.next.ID,
182 }
183
184 for _, addr := range it.next.Addrs {
185 rec.Addrs = append(rec.Addrs, types.Multiaddr{Multiaddr: addr})
186 }
187
188 return rec
189 }
190
191 func (it *peerChanIter) Close() error {
192 it.cancel()
193 return nil
194 }