Skip to content

Commit 8424283

Browse files
supriya-premkumarsupriya-premkumar
authored
V2 Retrieval Isolation (#1181)
1 parent 7ce439f commit 8424283

24 files changed

+281
-88
lines changed

‎api/clients/node_client.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ func (c client) GetBlobHeader(
4141
blobIndex uint32,
4242
) (*core.BlobHeader, *merkletree.Proof, error) {
4343
conn, err := grpc.NewClient(
44-
core.OperatorSocket(socket).GetRetrievalSocket(),
44+
core.OperatorSocket(socket).GetV1RetrievalSocket(),
4545
grpc.WithTransportCredentials(insecure.NewCredentials()),
4646
)
4747
if err != nil {
@@ -86,7 +86,7 @@ func (c client) GetChunks(
8686
chunksChan chan RetrievedChunks,
8787
) {
8888
conn, err := grpc.NewClient(
89-
core.OperatorSocket(opInfo.Socket).GetRetrievalSocket(),
89+
core.OperatorSocket(opInfo.Socket).GetV1RetrievalSocket(),
9090
grpc.WithTransportCredentials(insecure.NewCredentials()),
9191
)
9292
if err != nil {

‎api/clients/v2/retrieval_client.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ func (r *retrievalClient) getChunksFromOperator(
157157
chunksChan chan clients.RetrievedChunks,
158158
) {
159159
conn, err := grpc.NewClient(
160-
core.OperatorSocket(opInfo.Socket).GetRetrievalSocket(),
160+
core.OperatorSocket(opInfo.Socket).GetV2RetrievalSocket(),
161161
grpc.WithTransportCredentials(insecure.NewCredentials()),
162162
)
163163
defer func() {

‎core/mock/state.go

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ type PrivateOperatorInfo struct {
3232
DispersalPort string
3333
RetrievalPort string
3434
V2DispersalPort string
35+
V2RetrievalPort string
3536
}
3637

3738
type PrivateOperatorState struct {
@@ -140,7 +141,8 @@ func (d *ChainDataMock) GetTotalOperatorStateWithQuorums(ctx context.Context, bl
140141
dispersalPort := fmt.Sprintf("3%03v", 2*i)
141142
retrievalPort := fmt.Sprintf("3%03v", 2*i+1)
142143
v2DispersalPort := fmt.Sprintf("3%03v", 2*i+2)
143-
socket := core.MakeOperatorSocket(host, dispersalPort, retrievalPort, v2DispersalPort)
144+
v2RetrievalPort := fmt.Sprintf("3%03v", 2*i+3)
145+
socket := core.MakeOperatorSocket(host, dispersalPort, retrievalPort, v2DispersalPort, v2RetrievalPort)
144146

145147
indexed := &core.IndexedOperatorInfo{
146148
Socket: string(socket),
@@ -161,6 +163,7 @@ func (d *ChainDataMock) GetTotalOperatorStateWithQuorums(ctx context.Context, bl
161163
DispersalPort: dispersalPort,
162164
RetrievalPort: retrievalPort,
163165
V2DispersalPort: v2DispersalPort,
166+
V2RetrievalPort: v2RetrievalPort,
164167
}
165168

166169
indexedOperators[id] = indexed

‎core/serialization.go

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -528,25 +528,33 @@ func decode(data []byte, obj any) error {
528528
}
529529

530530
func (s OperatorSocket) GetV1DispersalSocket() string {
531-
ip, v1DispersalPort, _, _, err := ParseOperatorSocket(string(s))
531+
ip, v1DispersalPort, _, _, _, err := ParseOperatorSocket(string(s))
532532
if err != nil {
533533
return ""
534534
}
535535
return fmt.Sprintf("%s:%s", ip, v1DispersalPort)
536536
}
537537

538538
func (s OperatorSocket) GetV2DispersalSocket() string {
539-
ip, _, _, v2DispersalPort, err := ParseOperatorSocket(string(s))
539+
ip, _, _, v2DispersalPort, _, err := ParseOperatorSocket(string(s))
540540
if err != nil || v2DispersalPort == "" {
541541
return ""
542542
}
543543
return fmt.Sprintf("%s:%s", ip, v2DispersalPort)
544544
}
545545

546-
func (s OperatorSocket) GetRetrievalSocket() string {
547-
ip, _, retrievalPort, _, err := ParseOperatorSocket(string(s))
546+
func (s OperatorSocket) GetV1RetrievalSocket() string {
547+
ip, _, v1retrievalPort, _, _, err := ParseOperatorSocket(string(s))
548548
if err != nil {
549549
return ""
550550
}
551-
return fmt.Sprintf("%s:%s", ip, retrievalPort)
551+
return fmt.Sprintf("%s:%s", ip, v1retrievalPort)
552+
}
553+
554+
func (s OperatorSocket) GetV2RetrievalSocket() string {
555+
ip, _, _, _, v2RetrievalPort, err := ParseOperatorSocket(string(s))
556+
if err != nil || v2RetrievalPort == "" {
557+
return ""
558+
}
559+
return fmt.Sprintf("%s:%s", ip, v2RetrievalPort)
552560
}

‎core/serialization_test.go

Lines changed: 79 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -195,36 +195,37 @@ func TestHashPubKeyG1(t *testing.T) {
195195
}
196196

197197
func TestParseOperatorSocket(t *testing.T) {
198-
operatorSocket := "localhost:1234;5678;9999"
199-
host, dispersalPort, retrievalPort, v2DispersalPort, err := core.ParseOperatorSocket(operatorSocket)
198+
operatorSocket := "localhost:1234;5678;9999;10001"
199+
host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, v2RetrievalPort, err := core.ParseOperatorSocket(operatorSocket)
200200
assert.NoError(t, err)
201201
assert.Equal(t, "localhost", host)
202-
assert.Equal(t, "1234", dispersalPort)
203-
assert.Equal(t, "5678", retrievalPort)
202+
assert.Equal(t, "1234", v1DispersalPort)
203+
assert.Equal(t, "5678", v1RetrievalPort)
204204
assert.Equal(t, "9999", v2DispersalPort)
205+
assert.Equal(t, "10001", v2RetrievalPort)
205206

206-
host, dispersalPort, retrievalPort, v2DispersalPort, err = core.ParseOperatorSocket("localhost:1234;5678")
207+
host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, _, err = core.ParseOperatorSocket("localhost:1234;5678")
207208
assert.NoError(t, err)
208209
assert.Equal(t, "localhost", host)
209-
assert.Equal(t, "1234", dispersalPort)
210-
assert.Equal(t, "5678", retrievalPort)
210+
assert.Equal(t, "1234", v1DispersalPort)
211+
assert.Equal(t, "5678", v1RetrievalPort)
211212
assert.Equal(t, "", v2DispersalPort)
212213

213-
_, _, _, _, err = core.ParseOperatorSocket("localhost;1234;5678")
214+
_, _, _, _, _, err = core.ParseOperatorSocket("localhost;1234;5678")
214215
assert.NotNil(t, err)
215-
assert.ErrorContains(t, err, "invalid socket address format")
216+
assert.ErrorContains(t, err, "invalid host address format")
216217

217-
_, _, _, _, err = core.ParseOperatorSocket("localhost:12345678")
218+
_, _, _, _, _, err = core.ParseOperatorSocket("localhost:12345678")
218219
assert.NotNil(t, err)
219-
assert.ErrorContains(t, err, "invalid socket address format")
220+
assert.ErrorContains(t, err, "invalid v1 dispersal port format")
220221

221-
_, _, _, _, err = core.ParseOperatorSocket("localhost1234;5678")
222+
_, _, _, _, _, err = core.ParseOperatorSocket("localhost1234;5678")
222223
assert.NotNil(t, err)
223-
assert.ErrorContains(t, err, "invalid socket address format")
224+
assert.ErrorContains(t, err, "invalid host address format")
224225
}
225226

226227
func TestGetV1DispersalSocket(t *testing.T) {
227-
operatorSocket := core.OperatorSocket("localhost:1234;5678;9999")
228+
operatorSocket := core.OperatorSocket("localhost:1234;5678;9999;1025")
228229
socket := operatorSocket.GetV1DispersalSocket()
229230
assert.Equal(t, "localhost:1234", socket)
230231

@@ -234,28 +235,84 @@ func TestGetV1DispersalSocket(t *testing.T) {
234235

235236
operatorSocket = core.OperatorSocket("localhost:1234;5678;")
236237
socket = operatorSocket.GetV1DispersalSocket()
237-
assert.Equal(t, "localhost:1234", socket)
238+
assert.Equal(t, "", socket)
238239

239240
operatorSocket = core.OperatorSocket("localhost:1234")
240241
socket = operatorSocket.GetV1DispersalSocket()
241242
assert.Equal(t, "", socket)
242243
}
243244

244-
func TestGetRetrievalSocket(t *testing.T) {
245-
operatorSocket := core.OperatorSocket("localhost:1234;5678;9999")
246-
socket := operatorSocket.GetRetrievalSocket()
245+
func TestGetV1RetrievalSocket(t *testing.T) {
246+
// Valid v1/v2 socket
247+
operatorSocket := core.OperatorSocket("localhost:1234;5678;9999;10001")
248+
socket := operatorSocket.GetV1RetrievalSocket()
247249
assert.Equal(t, "localhost:5678", socket)
248250

251+
// Valid v1 socket
249252
operatorSocket = core.OperatorSocket("localhost:1234;5678")
250-
socket = operatorSocket.GetRetrievalSocket()
253+
socket = operatorSocket.GetV1RetrievalSocket()
251254
assert.Equal(t, "localhost:5678", socket)
252255

256+
// Invalid socket testcases
257+
operatorSocket = core.OperatorSocket("localhost:1234;5678;9999;10001;")
258+
socket = operatorSocket.GetV1RetrievalSocket()
259+
assert.Equal(t, "", socket)
260+
253261
operatorSocket = core.OperatorSocket("localhost:1234;5678;")
254-
socket = operatorSocket.GetRetrievalSocket()
255-
assert.Equal(t, "localhost:5678", socket)
262+
socket = operatorSocket.GetV1RetrievalSocket()
263+
assert.Equal(t, "", socket)
264+
265+
operatorSocket = core.OperatorSocket("localhost:;1234;5678;")
266+
socket = operatorSocket.GetV1RetrievalSocket()
267+
assert.Equal(t, "", socket)
268+
269+
operatorSocket = core.OperatorSocket("localhost:1234;:;5678;")
270+
socket = operatorSocket.GetV1RetrievalSocket()
271+
assert.Equal(t, "", socket)
272+
273+
operatorSocket = core.OperatorSocket("localhost:;;;")
274+
socket = operatorSocket.GetV1RetrievalSocket()
275+
assert.Equal(t, "", socket)
276+
277+
operatorSocket = core.OperatorSocket("localhost:1234")
278+
socket = operatorSocket.GetV1RetrievalSocket()
279+
assert.Equal(t, "", socket)
280+
}
281+
282+
func TestGetV2RetrievalSocket(t *testing.T) {
283+
// Valid v1/v2 socket
284+
operatorSocket := core.OperatorSocket("localhost:1234;5678;9999;10001")
285+
socket := operatorSocket.GetV2RetrievalSocket()
286+
assert.Equal(t, "localhost:10001", socket)
287+
288+
// Invalid v2 socket
289+
operatorSocket = core.OperatorSocket("localhost:1234;5678")
290+
socket = operatorSocket.GetV2RetrievalSocket()
291+
assert.Equal(t, "", socket)
292+
293+
// Invalid socket testcases
294+
operatorSocket = core.OperatorSocket("localhost:1234;5678;9999;10001;")
295+
socket = operatorSocket.GetV2RetrievalSocket()
296+
assert.Equal(t, "", socket)
297+
298+
operatorSocket = core.OperatorSocket("localhost:1234;5678;")
299+
socket = operatorSocket.GetV2RetrievalSocket()
300+
assert.Equal(t, "", socket)
301+
302+
operatorSocket = core.OperatorSocket("localhost:;1234;5678;")
303+
socket = operatorSocket.GetV2RetrievalSocket()
304+
assert.Equal(t, "", socket)
305+
306+
operatorSocket = core.OperatorSocket("localhost:1234;:;5678;")
307+
socket = operatorSocket.GetV2RetrievalSocket()
308+
assert.Equal(t, "", socket)
309+
310+
operatorSocket = core.OperatorSocket("localhost:;;;")
311+
socket = operatorSocket.GetV2RetrievalSocket()
312+
assert.Equal(t, "", socket)
256313

257314
operatorSocket = core.OperatorSocket("localhost:1234")
258-
socket = operatorSocket.GetRetrievalSocket()
315+
socket = operatorSocket.GetV2RetrievalSocket()
259316
assert.Equal(t, "", socket)
260317
}
261318

‎core/state.go

Lines changed: 50 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"encoding/json"
77
"fmt"
88
"math/big"
9+
"net"
910
"slices"
1011
"strings"
1112
)
@@ -19,48 +20,73 @@ func (s OperatorSocket) String() string {
1920
return string(s)
2021
}
2122

22-
func MakeOperatorSocket(nodeIP, dispersalPort, retrievalPort, v2DispersalPort string) OperatorSocket {
23-
if v2DispersalPort == "" {
23+
func MakeOperatorSocket(nodeIP, dispersalPort, retrievalPort, v2DispersalPort, v2RetrievalPort string) OperatorSocket {
24+
//TODO: Add config checks for invalid v1/v2 configs -- for v1, both v2 ports must be empty and for v2, both ports must be valid, reject any other combinations.
25+
if v2DispersalPort == "" && v2RetrievalPort == "" {
2426
return OperatorSocket(fmt.Sprintf("%s:%s;%s", nodeIP, dispersalPort, retrievalPort))
2527
}
26-
return OperatorSocket(fmt.Sprintf("%s:%s;%s;%s", nodeIP, dispersalPort, retrievalPort, v2DispersalPort))
28+
return OperatorSocket(fmt.Sprintf("%s:%s;%s;%s;%s", nodeIP, dispersalPort, retrievalPort, v2DispersalPort, v2RetrievalPort))
2729
}
2830

2931
type StakeAmount = *big.Int
3032

31-
func ParseOperatorSocket(socket string) (host string, dispersalPort string, retrievalPort string, v2DispersalPort string, err error) {
33+
func ParseOperatorSocket(socket string) (host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, v2RetrievalPort string, err error) {
34+
3235
s := strings.Split(socket, ";")
3336

34-
if len(s) == 2 {
35-
// no v2 dispersal port
36-
retrievalPort = s[1]
37-
s = strings.Split(s[0], ":")
38-
if len(s) != 2 {
39-
err = fmt.Errorf("invalid socket address format: %s", socket)
40-
return
41-
}
42-
host = s[0]
43-
dispersalPort = s[1]
37+
host, v1DispersalPort, err = net.SplitHostPort(s[0])
38+
if err != nil {
39+
err = fmt.Errorf("invalid host address format in %s: it must specify valid IP or host name (ex. 0.0.0.0:32004;32005;32006;32007)", socket)
4440

4541
return
4642
}
43+
if _, err = net.LookupHost(host); err != nil {
44+
//Invalid host
45+
host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, v2RetrievalPort, err =
46+
"", "", "", "", "",
47+
fmt.Errorf("invalid host address format in %s: it must specify valid IP or host name (ex. 0.0.0.0:32004;32005;32006;32007)", socket)
48+
return
49+
}
50+
if err = ValidatePort(v1DispersalPort); err != nil {
51+
host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, v2RetrievalPort, err =
52+
"", "", "", "", "",
53+
fmt.Errorf("invalid v1 dispersal port format in %s: it must specify valid v1 dispersal port (ex. 0.0.0.0:32004;32005;32006;32007)", socket)
54+
return
55+
}
4756

48-
if len(s) == 3 {
49-
// all ports specified
57+
switch len(s) {
58+
case 4:
5059
v2DispersalPort = s[2]
51-
retrievalPort = s[1]
60+
if err = ValidatePort(v2DispersalPort); err != nil {
61+
host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, v2RetrievalPort, err =
62+
"", "", "", "", "",
63+
fmt.Errorf("invalid v2 dispersal port format in %s: it must specify valid v2 dispersal port (ex. 0.0.0.0:32004;32005;32006;32007)", socket)
64+
return
65+
}
5266

53-
s = strings.Split(s[0], ":")
54-
if len(s) != 2 {
55-
err = fmt.Errorf("invalid socket address format: %s", socket)
67+
v2RetrievalPort = s[3]
68+
if err = ValidatePort(v2RetrievalPort); err != nil {
69+
host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, v2RetrievalPort, err =
70+
"", "", "", "", "",
71+
fmt.Errorf("invalid v2 retrieval port format in %s: it must specify valid v2 retrieval port (ex. 0.0.0.0:32004;32005;32006;32007)", socket)
5672
return
5773
}
58-
host = s[0]
59-
dispersalPort = s[1]
74+
fallthrough
75+
case 2:
76+
// V1 Parsing
77+
v1RetrievalPort = s[1]
78+
if err = ValidatePort(v1RetrievalPort); err != nil {
79+
host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, v2RetrievalPort, err =
80+
"", "", "", "", "",
81+
fmt.Errorf("invalid v1 retrieval port format in %s: it must specify valid v1 retrieval port (ex. 0.0.0.0:32004;32005;32006;32007)", socket)
82+
}
83+
return
84+
default:
85+
host, v1DispersalPort, v1RetrievalPort, v2DispersalPort, v2RetrievalPort, err =
86+
"", "", "", "", "",
87+
fmt.Errorf("invalid socket address format %s: it must specify v1 dispersal/retrieval ports, or v2 dispersal/retrieval ports (ex. 0.0.0.0:32004;32005;32006;32007)", socket)
6088
return
6189
}
62-
63-
return "", "", "", "", fmt.Errorf("invalid socket address format %s: it must specify dispersal port, retrieval port, and/or v2 dispersal port (ex. 0.0.0.0:32004;32005;32006)", socket)
6490
}
6591

6692
// OperatorInfo contains information about an operator which is stored on the blockchain state,

‎core/utils.go

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
11
package core
22

33
import (
4+
"fmt"
45
"math"
56
"math/big"
7+
"strconv"
68

79
"golang.org/x/exp/constraints"
810
)
@@ -23,3 +25,15 @@ func NextPowerOf2[T constraints.Integer](d T) T {
2325
nextPower := math.Ceil(math.Log2(float64(d)))
2426
return T(math.Pow(2.0, nextPower))
2527
}
28+
29+
func ValidatePort(portStr string) error {
30+
port, err := strconv.Atoi(portStr)
31+
if err != nil {
32+
return fmt.Errorf("port is not a valid number: %v", err)
33+
}
34+
35+
if port < 1 || port > 65535 {
36+
return fmt.Errorf("port number out of valid range (1-65535)")
37+
}
38+
return err
39+
}

‎disperser/common/semver/semver.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ func ScanOperators(operators map[core.OperatorID]*core.IndexedOperatorInfo, oper
3131
operatorSocket := core.OperatorSocket(operators[operatorId].Socket)
3232
var socket string
3333
if useRetrievalSocket {
34-
socket = operatorSocket.GetRetrievalSocket()
34+
socket = operatorSocket.GetV1RetrievalSocket()
3535
} else {
3636
socket = operatorSocket.GetV1DispersalSocket()
3737
}

‎disperser/controller/dispatcher.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,7 @@ func (d *Dispatcher) HandleBatch(ctx context.Context) (chan core.SigningMessage,
148148
for opID, op := range state.IndexedOperators {
149149
opID := opID
150150
op := op
151-
host, _, _, v2DispersalPort, err := core.ParseOperatorSocket(op.Socket)
151+
host, _, _, v2DispersalPort, _, err := core.ParseOperatorSocket(op.Socket)
152152
if err != nil {
153153
return nil, nil, fmt.Errorf("failed to parse operator socket (%s): %w", op.Socket, err)
154154
}

0 commit comments

Comments
 (0)