/
context.go
96 lines (87 loc) · 3 KB
/
context.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
package sync
import (
"github.com/libp2p/go-libp2p/core/network"
"github.com/libp2p/go-libp2p/core/protocol"
"github.com/pkg/errors"
"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/signing"
"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p"
"github.com/prysmaticlabs/prysm/v5/config/params"
)
// Specifies the fixed size context length.
const forkDigestLength = 4
// writes peer's current context for the expected payload to the stream.
func writeContextToStream(objCtx []byte, stream network.Stream) error {
// The rpc context for our v2 methods is the fork-digest of
// the relevant payload. We write the associated fork-digest(context)
// into the stream for the payload.
rpcCtx, err := expectRpcContext(stream)
if err != nil {
return err
}
// Exit early if an empty context is expected.
if !rpcCtx {
return nil
}
_, err = stream.Write(objCtx)
return err
}
// reads any attached context-bytes to the payload.
func readContextFromStream(stream network.Stream) ([]byte, error) {
hasCtx, err := expectRpcContext(stream)
if err != nil {
return nil, err
}
if !hasCtx {
return []byte{}, nil
}
// Read context (fork-digest) from stream
b := make([]byte, forkDigestLength)
if _, err := stream.Read(b); err != nil {
return nil, err
}
return b, nil
}
func expectRpcContext(stream network.Stream) (bool, error) {
_, message, version, err := p2p.TopicDeconstructor(string(stream.Protocol()))
if err != nil {
return false, err
}
// For backwards compatibility, we want to omit context bytes for certain v1 methods that were defined before
// context bytes were introduced into the protocol.
if version == p2p.SchemaVersionV1 && p2p.OmitContextBytesV1[message] {
return false, nil
}
return true, nil
}
// Minimal interface for a stream with a protocol.
type withProtocol interface {
Protocol() protocol.ID
}
// Validates that the rpc topic matches the provided version.
func validateVersion(version string, stream withProtocol) error {
_, _, streamVersion, err := p2p.TopicDeconstructor(string(stream.Protocol()))
if err != nil {
return err
}
if streamVersion != version {
return errors.Errorf("stream version of %s doesn't match provided version %s", streamVersion, version)
}
return nil
}
// ContextByteVersions is a mapping between expected values for context bytes
// and the runtime/version identifier they correspond to. This can be used to look up the type
// needed to unmarshal a wire-encoded value.
type ContextByteVersions map[[4]byte]int
// ContextByteVersionsForValRoot computes a mapping between all possible context bytes values
// and the runtime/version identifier for the corresponding fork.
func ContextByteVersionsForValRoot(valRoot [32]byte) (ContextByteVersions, error) {
m := make(ContextByteVersions)
for fv, v := range params.ConfigForkVersions(params.BeaconConfig()) {
digest, err := signing.ComputeForkDigest(fv[:], valRoot[:])
if err != nil {
return nil, errors.Wrapf(err, "unable to compute fork digest for fork version %#x", fv)
}
m[digest] = v
}
return m, nil
}