GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

mirrored from https://github.com/cloudflare/cloudflaredAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2023.4.2

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/control.go

128lines · modeblame

RawDownload
Latest commit unavailable.
12ad264eSudarsan Reddy4 years ago1package connection
2
3import (
4"context"
5"io"
dd540af6Devin Carr4 years ago6"net"
12ad264eSudarsan Reddy4 years ago7"time"
8
9"github.com/rs/zerolog"
10
991f01feDevin Carr3 years ago11"github.com/cloudflare/cloudflared/management"
12ad264eSudarsan Reddy4 years ago12tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
13)
14
15// RPCClientFunc derives a named tunnel rpc client that can then be used to register and unregister connections.
16type RPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient
17
18type controlStream struct {
19observer *Observer
20
e22422aacthuang4 years ago21connectedFuse ConnectedFuse
22namedTunnelProperties *NamedTunnelProperties
23connIndex uint8
dd540af6Devin Carr4 years ago24edgeAddress net.IP
99f39225Sudarsan Reddy3 years ago25protocol Protocol
12ad264eSudarsan Reddy4 years ago26
27newRPCClientFunc RPCClientFunc
28
29gracefulShutdownC <-chan struct{}
30gracePeriod time.Duration
31stoppedGracefully bool
32}
33
34// ControlStreamHandler registers connections with origintunneld and initiates graceful shutdown.
35type ControlStreamHandler interface {
1086d5edNuno Diegues4 years ago36// ServeControlStream handles the control plane of the transport in the current goroutine calling this
99d4e486João Oliveirinha4 years ago37ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions, tunnelConfigGetter TunnelConfigJSONGetter) error
1086d5edNuno Diegues4 years ago38// IsStopped tells whether the method above has finished
12ad264eSudarsan Reddy4 years ago39IsStopped() bool
40}
41
99d4e486João Oliveirinha4 years ago42type TunnelConfigJSONGetter interface {
43GetConfigJSON() ([]byte, error)
44}
45
12ad264eSudarsan Reddy4 years ago46// NewControlStream returns a new instance of ControlStreamHandler
47func NewControlStream(
48observer *Observer,
49connectedFuse ConnectedFuse,
e22422aacthuang4 years ago50namedTunnelConfig *NamedTunnelProperties,
12ad264eSudarsan Reddy4 years ago51connIndex uint8,
dd540af6Devin Carr4 years ago52edgeAddress net.IP,
12ad264eSudarsan Reddy4 years ago53newRPCClientFunc RPCClientFunc,
54gracefulShutdownC <-chan struct{},
55gracePeriod time.Duration,
99f39225Sudarsan Reddy3 years ago56protocol Protocol,
12ad264eSudarsan Reddy4 years ago57) ControlStreamHandler {
58if newRPCClientFunc == nil {
59newRPCClientFunc = newRegistrationRPCClient
60}
61return &controlStream{
e22422aacthuang4 years ago62observer: observer,
63connectedFuse: connectedFuse,
64namedTunnelProperties: namedTunnelConfig,
65newRPCClientFunc: newRPCClientFunc,
66connIndex: connIndex,
dd540af6Devin Carr4 years ago67edgeAddress: edgeAddress,
e22422aacthuang4 years ago68gracefulShutdownC: gracefulShutdownC,
69gracePeriod: gracePeriod,
99f39225Sudarsan Reddy3 years ago70protocol: protocol,
12ad264eSudarsan Reddy4 years ago71}
72}
73
74func (c *controlStream) ServeControlStream(
75ctx context.Context,
76rw io.ReadWriteCloser,
77connOptions *tunnelpogs.ConnectionOptions,
99d4e486João Oliveirinha4 years ago78tunnelConfigGetter TunnelConfigJSONGetter,
12ad264eSudarsan Reddy4 years ago79) error {
80rpcClient := c.newRPCClientFunc(ctx, rw, c.observer.log)
81
dd540af6Devin Carr4 years ago82registrationDetails, err := rpcClient.RegisterConnection(ctx, c.namedTunnelProperties, connOptions, c.connIndex, c.edgeAddress, c.observer)
99d4e486João Oliveirinha4 years ago83if err != nil {
d7da74cbSudarsan Reddy4 years ago84rpcClient.Close()
12ad264eSudarsan Reddy4 years ago85return err
86}
99f39225Sudarsan Reddy3 years ago87
991f01feDevin Carr3 years ago88c.observer.logConnected(registrationDetails.UUID, c.connIndex, registrationDetails.Location, c.edgeAddress, c.protocol)
99f39225Sudarsan Reddy3 years ago89c.observer.sendConnectedEvent(c.connIndex, c.protocol, registrationDetails.Location)
12ad264eSudarsan Reddy4 years ago90c.connectedFuse.Connected()
91
99d4e486João Oliveirinha4 years ago92// if conn index is 0 and tunnel is not remotely managed, then send local ingress rules configuration
93if c.connIndex == 0 && !registrationDetails.TunnelIsRemotelyManaged {
94if tunnelConfig, err := tunnelConfigGetter.GetConfigJSON(); err == nil {
95if err := rpcClient.SendLocalConfiguration(ctx, tunnelConfig, c.observer); err != nil {
96c.observer.log.Err(err).Msg("unable to send local configuration")
97}
98} else {
99c.observer.log.Err(err).Msg("failed to obtain current configuration")
100}
101}
102
1086d5edNuno Diegues4 years ago103c.waitForUnregister(ctx, rpcClient)
27e1277aSudarsan Reddy4 years ago104return nil
105}
106
107func (c *controlStream) waitForUnregister(ctx context.Context, rpcClient NamedTunnelRPCClient) {
12ad264eSudarsan Reddy4 years ago108// wait for connection termination or start of graceful shutdown
d7da74cbSudarsan Reddy4 years ago109defer rpcClient.Close()
12ad264eSudarsan Reddy4 years ago110select {
111case <-ctx.Done():
112break
113case <-c.gracefulShutdownC:
114c.stoppedGracefully = true
115}
116
117c.observer.sendUnregisteringEvent(c.connIndex)
118rpcClient.GracefulShutdown(ctx, c.gracePeriod)
991f01feDevin Carr3 years ago119c.observer.log.Info().
120Int(management.EventTypeKey, int(management.Cloudflared)).
121Uint8(LogFieldConnIndex, c.connIndex).
122IPAddr(LogFieldIPAddress, c.edgeAddress).
123Msg("Unregistered tunnel connection")
12ad264eSudarsan Reddy4 years ago124}
125
126func (c *controlStream) IsStopped() bool {
127return c.stoppedGracefully
128}