-
Notifications
You must be signed in to change notification settings - Fork 51
/
enforcerproxy.go
275 lines (229 loc) · 8.27 KB
/
enforcerproxy.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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
// Package enforcerproxy :: This is the implementation of the RPC client
// It implements the interface of Trireme Enforcer and forwards these
// requests to the actual remote enforcer instead of implementing locally
package enforcerproxy
import (
"errors"
"fmt"
"sync"
"time"
"go.uber.org/zap"
"github.com/aporeto-inc/trireme/collector"
"github.com/aporeto-inc/trireme/constants"
"github.com/aporeto-inc/trireme/crypto"
"github.com/aporeto-inc/trireme/enforcer"
"github.com/aporeto-inc/trireme/enforcer/utils/fqconfig"
"github.com/aporeto-inc/trireme/enforcer/utils/rpcwrapper"
"github.com/aporeto-inc/trireme/enforcer/utils/secrets"
"github.com/aporeto-inc/trireme/policy"
"github.com/aporeto-inc/trireme/processmon"
)
//keyPEM is a private interface required by the enforcerlauncher to expose method not exposed by the
//PolicyEnforcer interface
type keyPEM interface {
AuthPEM() []byte
TransmittedPEM() []byte
EncodingPEM() []byte
}
//ErrFailedtoLaunch exported
var ErrFailedtoLaunch = errors.New("Failed to Launch")
//ErrExpectedEnforcer exported
var ErrExpectedEnforcer = errors.New("Process was not launched")
// ErrEnforceFailed exported
var ErrEnforceFailed = errors.New("Failed to enforce rules")
// ErrInitFailed exported
var ErrInitFailed = errors.New("Failed remote Init")
//ProxyInfo is the struct used to hold state about active enforcers in the system
type ProxyInfo struct {
MutualAuth bool
Secrets secrets.Secrets
serverID string
validity time.Duration
prochdl processmon.ProcessManager
rpchdl rpcwrapper.RPCClient
initDone map[string]bool
filterQueue *fqconfig.FilterQueue
commandArg string
statsServerSecret string
procMountPoint string
externalIPCacheTimeout time.Duration
sync.Mutex
}
//InitRemoteEnforcer method makes a RPC call to the remote enforcer
func (s *ProxyInfo) InitRemoteEnforcer(contextID string) error {
resp := &rpcwrapper.Response{}
request := &rpcwrapper.Request{
Payload: &rpcwrapper.InitRequestPayload{
FqConfig: s.filterQueue,
MutualAuth: s.MutualAuth,
Validity: s.validity,
SecretType: s.Secrets.Type(),
ServerID: s.serverID,
CAPEM: s.Secrets.(keyPEM).AuthPEM(),
PublicPEM: s.Secrets.(keyPEM).TransmittedPEM(),
PrivatePEM: s.Secrets.(keyPEM).EncodingPEM(),
ExternalIPCacheTimeout: s.externalIPCacheTimeout,
},
}
if s.Secrets.Type() == secrets.PKICompactType {
request.Payload.(*rpcwrapper.InitRequestPayload).Token = s.Secrets.TransmittedKey()
}
if err := s.rpchdl.RemoteCall(contextID, "Server.InitEnforcer", request, resp); err != nil {
return fmt.Errorf("Failed to initialize remote enforcer: status %s, error: %s", resp.Status, err.Error())
}
s.Lock()
s.initDone[contextID] = true
s.Unlock()
return nil
}
//Enforce method makes a RPC call for the remote enforcer enforce method
func (s *ProxyInfo) Enforce(contextID string, puInfo *policy.PUInfo) error {
zap.L().Debug("PID of container", zap.Int("pid", puInfo.Runtime.Pid()))
zap.L().Debug("NSPath of container", zap.String("ns", puInfo.Runtime.NSPath()))
err := s.prochdl.LaunchProcess(contextID, puInfo.Runtime.Pid(), puInfo.Runtime.NSPath(), s.rpchdl, s.commandArg, s.statsServerSecret, s.procMountPoint)
if err != nil {
return err
}
zap.L().Debug("Called enforce and launched process", zap.String("contextID", contextID))
s.Lock()
_, ok := s.initDone[contextID]
s.Unlock()
if !ok {
if err = s.InitRemoteEnforcer(contextID); err != nil {
return err
}
}
request := &rpcwrapper.Request{
Payload: &rpcwrapper.EnforcePayload{
ContextID: contextID,
ManagementID: puInfo.Policy.ManagementID(),
TriremeAction: puInfo.Policy.TriremeAction(),
ApplicationACLs: puInfo.Policy.ApplicationACLs(),
NetworkACLs: puInfo.Policy.NetworkACLs(),
PolicyIPs: puInfo.Policy.IPAddresses(),
Annotations: puInfo.Policy.Annotations(),
Identity: puInfo.Policy.Identity(),
ReceiverRules: puInfo.Policy.ReceiverRules(),
TransmitterRules: puInfo.Policy.TransmitterRules(),
TriremeNetworks: puInfo.Policy.TriremeNetworks(),
ExcludedNetworks: puInfo.Policy.ExcludedNetworks(),
},
}
err = s.rpchdl.RemoteCall(contextID, "Server.Enforce", request, &rpcwrapper.Response{})
if err != nil {
// We can't talk to the enforcer. Kill it and restart it
s.Lock()
delete(s.initDone, contextID)
s.Unlock()
s.prochdl.KillProcess(contextID)
zap.L().Error("Failed to Enforce remote enforcer", zap.Error(err))
return ErrEnforceFailed
}
return nil
}
// Unenforce stops enforcing policy for the given contextID.
func (s *ProxyInfo) Unenforce(contextID string) error {
s.Lock()
delete(s.initDone, contextID)
s.Unlock()
return nil
}
// GetFilterQueue returns the current FilterQueueConfig.
func (s *ProxyInfo) GetFilterQueue() *fqconfig.FilterQueue {
return s.filterQueue
}
// Start starts the the remote enforcer proxy.
func (s *ProxyInfo) Start() error {
return nil
}
// Stop stops the remote enforcer.
func (s *ProxyInfo) Stop() error {
return nil
}
//NewProxyEnforcer creates a new proxy to remote enforcers
func NewProxyEnforcer(mutualAuth bool,
filterQueue *fqconfig.FilterQueue,
collector collector.EventCollector,
service enforcer.PacketProcessor,
secrets secrets.Secrets,
serverID string,
validity time.Duration,
rpchdl rpcwrapper.RPCClient,
cmdArg string,
procMountPoint string,
externalIPCacheTimeout time.Duration,
) enforcer.PolicyEnforcer {
statsServersecret, err := crypto.GenerateRandomString(32)
if err != nil {
// There is a very small chance of this happening we will log an error here.
zap.L().Error("Failed to generate random secret for stats reporting.Falling back to static secret", zap.Error(err))
// We will use current time as the secret
statsServersecret = time.Now().String()
}
proxydata := &ProxyInfo{
MutualAuth: mutualAuth,
Secrets: secrets,
serverID: serverID,
validity: validity,
prochdl: processmon.GetProcessManagerHdl(),
rpchdl: rpchdl,
initDone: make(map[string]bool),
filterQueue: filterQueue,
commandArg: cmdArg,
statsServerSecret: statsServersecret,
procMountPoint: procMountPoint,
externalIPCacheTimeout: externalIPCacheTimeout,
}
zap.L().Debug("Called NewDataPathEnforcer")
statsServer := rpcwrapper.NewRPCWrapper()
rpcServer := &StatsServer{rpchdl: statsServer, collector: collector, secret: statsServersecret}
// Start hte server for statistics collection
go statsServer.StartServer("unix", rpcwrapper.StatsChannel, rpcServer) // nolint
return proxydata
}
// NewDefaultProxyEnforcer This is the default datapth method. THis is implemented to keep the interface consistent whether we are local or remote enforcer
func NewDefaultProxyEnforcer(serverID string,
collector collector.EventCollector,
secrets secrets.Secrets,
rpchdl rpcwrapper.RPCClient,
procMountPoint string,
) enforcer.PolicyEnforcer {
mutualAuthorization := false
fqConfig := fqconfig.NewFilterQueueWithDefaults()
defaultExternalIPCacheTimeout, err := time.ParseDuration(enforcer.DefaultExternalIPTimeout)
if err != nil {
defaultExternalIPCacheTimeout = time.Second
}
validity := time.Hour * 8760
return NewProxyEnforcer(
mutualAuthorization,
fqConfig,
collector,
nil,
secrets,
serverID,
validity,
rpchdl,
constants.DefaultRemoteArg,
procMountPoint,
defaultExternalIPCacheTimeout,
)
}
//StatsServer This struct is a receiver for Statsserver and maintains a handle to the RPC StatsServer
type StatsServer struct {
collector collector.EventCollector
rpchdl rpcwrapper.RPCServer
secret string
}
//GetStats is the function called from the remoteenforcer when it has new flow events to publish
func (r *StatsServer) GetStats(req rpcwrapper.Request, resp *rpcwrapper.Response) error {
if !r.rpchdl.ProcessMessage(&req, r.secret) {
zap.L().Error("Message sender cannot be verified")
return errors.New("Message sender cannot be verified")
}
payload := req.Payload.(rpcwrapper.StatsPayload)
for _, record := range payload.Flows {
r.collector.CollectFlowEvent(record)
}
return nil
}