forked from libp2p/go-libp2p-conn
/
conn.go
168 lines (138 loc) · 3.98 KB
/
conn.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
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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
package conn
import (
"context"
"errors"
"io"
"net"
logging "github.com/ipfs/go-log"
ci "github.com/libp2p/go-libp2p-crypto"
ic "github.com/libp2p/go-libp2p-crypto"
iconn "github.com/libp2p/go-libp2p-interface-conn"
lgbl "github.com/libp2p/go-libp2p-loggables"
peer "github.com/libp2p/go-libp2p-peer"
secio "github.com/libp2p/go-libp2p-secio"
tpt "github.com/libp2p/go-libp2p-transport"
smux "github.com/libp2p/go-stream-muxer"
ma "github.com/multiformats/go-multiaddr"
)
var log = logging.Logger("conn")
// singleConn represents a single stream-multipexed connection to another Peer (IPFS Node).
type singleConn struct {
streamConn smux.Conn
tptConn tpt.Conn
secSession secio.Session
event io.Closer
}
var _ iconn.Conn = &singleConn{}
// newSingleConn constructs a new connection
func newSingleConn(ctx context.Context, local, remote peer.ID, privKey ci.PrivKey, tptConn tpt.Conn, pstpt smux.Transport, isServer bool) (iconn.Conn, error) {
ml := lgbl.Dial("conn", local, remote, tptConn.LocalMultiaddr(), tptConn.RemoteMultiaddr())
var streamConn smux.Conn
var secSession secio.Session
c := tptConn
// 1. secure the connection
if privKey != nil && iconn.EncryptConnections {
var err error
secSession, err = setupSecureSession(ctx, local, privKey, tptConn)
if err != nil {
return nil, err
}
c = &secureConn{
insecure: tptConn,
secure: secSession,
}
} else {
log.Warning("creating INSECURE connection %s at %s", tptConn.LocalMultiaddr(), tptConn.RemoteMultiaddr())
}
// 2. start stream multipling
var err error
streamConn, err = pstpt.NewConn(c, isServer)
if err != nil {
return nil, err
}
sconn := &singleConn{
streamConn: streamConn,
tptConn: tptConn,
secSession: secSession,
event: log.EventBegin(ctx, "connLifetime", ml),
}
log.Debugf("newSingleConn %p: %v to %v", sconn, local, remote)
return sconn, nil
}
func setupSecureSession(ctx context.Context, local peer.ID, privKey ci.PrivKey, ch io.ReadWriteCloser) (secio.Session, error) {
if local == "" {
return nil, errors.New("local peer is nil")
}
if privKey == nil {
return nil, errors.New("private key is nil")
}
sessgen := secio.SessionGenerator{
LocalID: local,
PrivateKey: privKey,
}
return sessgen.NewSession(ctx, ch)
}
// close is the internal close function, called by ContextCloser.Close
func (c *singleConn) Close() error {
defer func() {
if c.event != nil {
c.event.Close()
c.event = nil
}
}()
// closing the stream muxer also closes the underlying net.Conn
return c.streamConn.Close()
}
// ID is an identifier unique to this connection.
func (c *singleConn) ID() string {
return iconn.ID(c)
}
func (c *singleConn) String() string {
return iconn.String(c, "singleConn")
}
func (c *singleConn) LocalAddr() net.Addr {
return c.tptConn.LocalAddr()
}
func (c *singleConn) RemoteAddr() net.Addr {
return c.tptConn.RemoteAddr()
}
func (c *singleConn) LocalPrivateKey() ic.PrivKey {
if c.secSession != nil {
return c.secSession.LocalPrivateKey()
}
return nil
}
func (c *singleConn) RemotePublicKey() ic.PubKey {
if c.secSession != nil {
return c.secSession.RemotePublicKey()
}
return nil
}
// LocalMultiaddr is the Multiaddr on this side
func (c *singleConn) LocalMultiaddr() ma.Multiaddr {
return c.tptConn.LocalMultiaddr()
}
// RemoteMultiaddr is the Multiaddr on the remote side
func (c *singleConn) RemoteMultiaddr() ma.Multiaddr {
return c.tptConn.RemoteMultiaddr()
}
func (c *singleConn) Transport() tpt.Transport {
return c.tptConn.Transport()
}
// LocalPeer is the Peer on this side
func (c *singleConn) LocalPeer() peer.ID {
return c.secSession.LocalPeer()
}
// RemotePeer is the Peer on the remote side
func (c *singleConn) RemotePeer() peer.ID {
return c.secSession.RemotePeer()
}
func (c *singleConn) AcceptStream() (smux.Stream, error) {
return c.streamConn.AcceptStream()
}
func (c *singleConn) OpenStream() (smux.Stream, error) {
return c.streamConn.OpenStream()
}
func (c *singleConn) IsClosed() bool {
return c.streamConn.IsClosed()
}