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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
package sidechannel
import (
"context"
"encoding/binary"
"fmt"
"io"
"net"
"strconv"
"time"
"github.com/sirupsen/logrus"
"gitlab.com/gitlab-org/gitaly/v14/internal/backchannel"
"gitlab.com/gitlab-org/gitaly/v14/internal/gitaly/client"
"gitlab.com/gitlab-org/gitaly/v14/internal/listenmux"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/metadata"
)
var magicBytes = []byte("sidechannel")
// sidechannelTimeout is the timeout for establishing a sidechannel
// connection. The sidechannel is supposed to be opened on the same wire with
// incoming grpc request. There won't be real handshaking involved, so it
// should be fast.
const (
sidechannelTimeout = 5 * time.Second
sidechannelMetadataKey = "gitaly-sidechannel-id"
)
// OpenSidechannel opens a sidechannel connection from the stream opener
// extracted from the current peer connection.
func OpenSidechannel(ctx context.Context) (_ *ServerConn, err error) {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil, fmt.Errorf("sidechannel: failed to extract incoming metadata")
}
ids := md.Get(sidechannelMetadataKey)
if len(ids) == 0 {
return nil, fmt.Errorf("sidechannel: sidechannel-id not found in incoming metadata")
}
sidechannelID, _ := strconv.ParseInt(ids[len(ids)-1], 10, 64)
muxSession, err := backchannel.GetYamuxSession(ctx)
if err != nil {
return nil, fmt.Errorf("sidechannel: fail to extract yamux session: %w", err)
}
stream, err := muxSession.Open()
if err != nil {
return nil, fmt.Errorf("sidechannel: open stream: %w", err)
}
defer func() {
if err != nil {
stream.Close()
}
}()
if err := stream.SetDeadline(time.Now().Add(sidechannelTimeout)); err != nil {
return nil, err
}
if _, err := stream.Write(magicBytes); err != nil {
return nil, fmt.Errorf("sidechannel: write magic bytes: %w", err)
}
if err := binary.Write(stream, binary.BigEndian, sidechannelID); err != nil {
return nil, fmt.Errorf("sidechannel: write stream id: %w", err)
}
buf := make([]byte, 2)
if _, err := io.ReadFull(stream, buf); err != nil {
return nil, fmt.Errorf("sidechannel: receive confirmation: %w", err)
}
if string(buf) != "ok" {
return nil, fmt.Errorf("sidechannel: expected ok, got %q", buf)
}
if err := stream.SetDeadline(time.Time{}); err != nil {
return nil, err
}
return newServerConn(stream), nil
}
// RegisterSidechannel registers the caller into the waiting list of the
// sidechannel registry and injects the sidechannel ID into outgoing metadata.
// The caller is expected to establish the request with the returned context. The
// callback is executed automatically when the sidechannel connection arrives.
// The result is pushed to the error channel of the returned waiter.
func RegisterSidechannel(ctx context.Context, registry *Registry, callback func(*ClientConn) error) (context.Context, *Waiter) {
waiter := registry.Register(callback)
ctxOut := metadata.AppendToOutgoingContext(ctx, sidechannelMetadataKey, fmt.Sprintf("%d", waiter.id))
return ctxOut, waiter
}
// ServerHandshaker implements the server-side sidechannel handshake.
type ServerHandshaker struct {
registry *Registry
}
// Magic returns the magic bytes for sidechannel
func (s *ServerHandshaker) Magic() string {
return string(magicBytes)
}
// Handshake implements the handshaking logic for sidechannel so that
// this handshaker reads the sidechannel ID from the wire, and then delegates
// the connection to the sidechannel registry
func (s *ServerHandshaker) Handshake(conn net.Conn, authInfo credentials.AuthInfo) (net.Conn, credentials.AuthInfo, error) {
var sidechannelID sidechannelID
if err := binary.Read(conn, binary.BigEndian, &sidechannelID); err != nil {
return nil, nil, fmt.Errorf("sidechannel: fail to extract sidechannel ID: %w", err)
}
if err := s.registry.receive(sidechannelID, conn); err != nil {
return nil, nil, err
}
// credentials.ErrConnDispatched, indicating that the connection is already
// dispatched out of gRPC. gRPC should leave it alone and exit in peace.
return nil, nil, credentials.ErrConnDispatched
}
// NewServerHandshaker creates a new handshaker for sidechannel to
// embed into listenmux.
func NewServerHandshaker(registry *Registry) *ServerHandshaker {
return &ServerHandshaker{registry: registry}
}
// NewClientHandshaker is used to enable sidechannel support on outbound
// gRPC connections.
func NewClientHandshaker(logger *logrus.Entry, registry *Registry) client.Handshaker {
return backchannel.NewClientHandshaker(
logger,
func() backchannel.Server {
lm := listenmux.New(insecure.NewCredentials())
lm.Register(NewServerHandshaker(registry))
return grpc.NewServer(grpc.Creds(lm))
},
)
}
|
