Merge pull request #1162 from safchain/auth-by-key

split agent / analyzer and api auth

Author: lebauce

agent/agent.go

@@ -71,7 +71,7 @@ func NewAnalyzerWSStructClientPool(authOptions *shttp.AuthenticationOpts) (*shtt
 
 	addresses, err := config.GetAnalyzerServiceAddresses()
 	if err != nil {
-		return nil, fmt.Errorf("Unable to get the analyzers list: %s", err.Error())
+		return nil, fmt.Errorf("Unable to get the analyzers list: %s", err)
 	}
 
 	if len(addresses) == 0 {
@@ -80,8 +80,7 @@ func NewAnalyzerWSStructClientPool(authOptions *shttp.AuthenticationOpts) (*shtt
 	}
 
 	for _, sa := range addresses {
-		authClient := shttp.NewAuthenticationClient(config.GetURL("http", sa.Addr, sa.Port, ""), authOptions)
-		c := shttp.NewWSClientFromConfig(common.AgentService, config.GetURL("ws", sa.Addr, sa.Port, "/ws/agent"), authClient, nil)
+		c := shttp.NewWSClientFromConfig(common.AgentService, config.GetURL("ws", sa.Addr, sa.Port, "/ws/agent"), authOptions, nil)
 		pool.AddClient(c)
 	}
 
@@ -172,20 +171,28 @@ func NewAgent() (*Agent, error) {
 	tm := topology.NewTIDMapper(g)
 	tm.Start()
 
+	apiAuthBackendName := config.GetString("agent.auth.api.backend")
+	apiAuthBackend, err := shttp.NewAuthenticationBackendByName(apiAuthBackendName)
+	if err != nil {
+		return nil, err
+	}
+
 	hserver, err := shttp.NewServerFromConfig(common.AgentService)
 	if err != nil {
 		return nil, err
 	}
 
+	hserver.RegisterLoginRoute(apiAuthBackend)
+
 	if err := hserver.Listen(); err != nil {
 		return nil, err
 	}
 
-	if _, err = api.NewAPI(hserver, nil, common.AgentService); err != nil {
+	if _, err = api.NewAPI(hserver, nil, common.AgentService, apiAuthBackend); err != nil {
 		return nil, err
 	}
 
-	wsServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/subscriber"))
+	wsServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/subscriber", apiAuthBackend))
 
 	// declare all extension available throught API and filtering
 	tr := traversal.NewGremlinTraversalParser()
@@ -198,13 +205,16 @@ func NewAgent() (*Agent, error) {
 		return nil, err
 	}
 
-	api.RegisterTopologyAPI(hserver, g, tr)
+	api.RegisterTopologyAPI(hserver, g, tr, apiAuthBackend)
 
-	authOptions := analyzer.NewAnalyzerAuthenticationOpts()
+	clusterAuthOptions := &shttp.AuthenticationOpts{
+		Username: config.GetString("agent.auth.cluster.username"),
+		Password: config.GetString("agent.auth.cluster.password"),
+	}
 
 	topologyEndpoint := topology.NewTopologySubscriberEndpoint(wsServer, g, tr)
 
-	analyzerClientPool, err := NewAnalyzerWSStructClientPool(authOptions)
+	analyzerClientPool, err := NewAnalyzerWSStructClientPool(clusterAuthOptions)
 	if err != nil {
 		return nil, err
 	}
@@ -239,7 +249,7 @@ func NewAgent() (*Agent, error) {
 
 	onDemandProbeServer, err := ondemand.NewOnDemandProbeServer(flowProbeBundle, g, analyzerClientPool)
 	if err != nil {
-		return nil, fmt.Errorf("Unable to initialize on-demand flow probe %s", err.Error())
+		return nil, fmt.Errorf("Unable to initialize on-demand flow probe %s", err)
 	}
 
 	agent := &Agent{
@@ -259,7 +269,7 @@ func NewAgent() (*Agent, error) {
 		topologyForwarder:   tforwarder,
 	}
 
-	api.RegisterStatusAPI(hserver, agent)
+	api.RegisterStatusAPI(hserver, agent, apiAuthBackend)
 
 	return agent, nil
 }

analyzer/flow_client.go

@@ -27,7 +27,6 @@ import (
 	"math/rand"
 	"net"
 	"net/url"
-	"strconv"
 	"strings"
 	"time"
 
@@ -108,11 +107,8 @@ func (c *FlowClientWebSocketConn) Close() error {
 
 // Connect to the WebSocket flow server
 func (c *FlowClientWebSocketConn) Connect() error {
-	authOptions := NewAnalyzerAuthenticationOpts()
-	authAddr := common.NormalizeAddrForURL(c.url.Hostname())
-	authPort, _ := strconv.Atoi(c.url.Port())
-	authClient := shttp.NewAuthenticationClient(config.GetURL("http", authAddr, authPort, ""), authOptions)
-	c.wsClient = shttp.NewWSClientFromConfig(common.AgentService, c.url, authClient, nil)
+	authOptions := AnalyzerClusterAuthenticationOpts()
+	c.wsClient = shttp.NewWSClientFromConfig(common.AgentService, c.url, authOptions, nil)
 	c.wsClient.Connect()
 	c.wsClient.AddEventHandler(c)
 

analyzer/flow_server.go

@@ -82,6 +82,7 @@ type FlowServerWebSocketConn struct {
 	timeOfLastLostFlowsLog time.Time
 	numOfLostFlows         int
 	maxFlowBufferSize      int
+	auth                   shttp.AuthenticationBackend
 }
 
 // FlowServer describes a flow server with pipeline enhancers mechanism
@@ -96,6 +97,7 @@ type FlowServer struct {
 	bulkInsertDeadline     time.Duration
 	ch                     chan *flow.Flow
 	quit                   chan struct{}
+	auth                   shttp.AuthenticationBackend
 }
 
 // OnMessage event
@@ -122,7 +124,7 @@ func (c *FlowServerWebSocketConn) OnMessage(client shttp.WSSpeaker, m shttp.WSMe
 // Serve starts a WebSocket flow server
 func (c *FlowServerWebSocketConn) Serve(ch chan *flow.Flow, quit chan struct{}, wg *sync.WaitGroup) {
 	c.ch = ch
-	server := shttp.NewWSServer(c.server, "/ws/flow")
+	server := shttp.NewWSServer(c.server, "/ws/flow", c.auth)
 	server.AddEventHandler(c)
 	go func() {
 		server.Start()
@@ -132,9 +134,9 @@ func (c *FlowServerWebSocketConn) Serve(ch chan *flow.Flow, quit chan struct{},
 }
 
 // NewFlowServerWebSocketConn returns a new WebSocket flow server
-func NewFlowServerWebSocketConn(server *shttp.Server) (*FlowServerWebSocketConn, error) {
+func NewFlowServerWebSocketConn(server *shttp.Server, auth shttp.AuthenticationBackend) (*FlowServerWebSocketConn, error) {
 	flowsMax := config.GetConfig().GetInt("analyzer.flow.max_buffer_size")
-	return &FlowServerWebSocketConn{server: server, maxFlowBufferSize: flowsMax}, nil
+	return &FlowServerWebSocketConn{server: server, maxFlowBufferSize: flowsMax, auth: auth}, nil
 }
 
 // Serve UDP connections
@@ -273,23 +275,23 @@ func (s *FlowServer) setupBulkConfigFromBackend() error {
 }
 
 // NewFlowServer creates a new flow server listening at address/port, based on configuration
-func NewFlowServer(s *shttp.Server, g *graph.Graph, store storage.Storage, probe *probe.ProbeBundle) (*FlowServer, error) {
+func NewFlowServer(s *shttp.Server, g *graph.Graph, store storage.Storage, probe *probe.ProbeBundle, auth shttp.AuthenticationBackend) (*FlowServer, error) {
 	pipeline := flow.NewEnhancerPipeline(enhancers.NewGraphFlowEnhancer(g))
 
 	// check that the neutron probe is loaded if so add the neutron flow enhancer
 	if probe.GetProbe("neutron") != nil {
 		pipeline.AddEnhancer(enhancers.NewNeutronFlowEnhancer(g))
 	}
 
-	var err error
 	var conn FlowServerConn
 	protocol := strings.ToLower(config.GetString("flow.protocol"))
 
+	var err error
 	switch protocol {
 	case "udp":
 		conn, err = NewFlowServerUDPConn(s.Addr, s.Port)
 	case "websocket":
-		conn, err = NewFlowServerWebSocketConn(s)
+		conn, err = NewFlowServerWebSocketConn(s, auth)
 	default:
 		err = fmt.Errorf("Invalid protocol %s", protocol)
 	}
@@ -304,6 +306,7 @@ func NewFlowServer(s *shttp.Server, g *graph.Graph, store storage.Storage, probe
 		enhancerPipelineConfig: flow.NewEnhancerPipelineConfig(),
 		conn: conn,
 		quit: make(chan struct{}, 2),
+		auth: auth,
 	}
 	err = fs.setupBulkConfigFromBackend()
 	if err != nil {

analyzer/server.go

@@ -76,12 +76,13 @@ func (s *Server) GetStatus() interface{} {
 		Incomers: make(map[string]shttp.WSConnStatus),
 		Outgoers: make(map[string]shttp.WSConnStatus),
 	}
-	for host, peer := range s.replicationEndpoint.conns {
-		if host == peer.GetHost() {
-			peersStatus.Incomers[host] = peer.GetStatus()
-		} else {
-			peersStatus.Outgoers[host] = peer.GetStatus()
-		}
+
+	for _, speaker := range s.replicationEndpoint.in.GetSpeakers() {
+		peersStatus.Incomers[speaker.GetRemoteHost()] = speaker.GetStatus()
+	}
+
+	for _, speaker := range s.replicationEndpoint.out.GetSpeakers() {
+		peersStatus.Outgoers[speaker.GetRemoteHost()] = speaker.GetStatus()
 	}
 
 	return &types.AnalyzerStatus{
@@ -190,7 +191,7 @@ func NewServerFromConfig() (*Server, error) {
 	for {
 		host := config.GetString("host_id")
 		if err = etcdClient.SetInt64(fmt.Sprintf("/analyzer:%s/start-time", host), time.Now().Unix()); err != nil {
-			logging.GetLogger().Errorf("Etcd server not ready: %s", err.Error())
+			logging.GetLogger().Errorf("Etcd server not ready: %s", err)
 			time.Sleep(time.Second)
 		} else {
 			break
@@ -228,83 +229,97 @@ func NewServerFromConfig() (*Server, error) {
 
 	g := graph.NewGraphFromConfig(cached, common.AnalyzerService)
 
-	authOptions := NewAnalyzerAuthenticationOpts()
+	clusterAuthOptions := AnalyzerClusterAuthenticationOpts()
+
+	clusterAuthBackendName := config.GetString("analyzer.auth.cluster.backend")
+	clusterAuthBackend, err := shttp.NewAuthenticationBackendByName(clusterAuthBackendName)
+	if err != nil {
+		return nil, err
+	}
+	// force admin user for the cluster backend to ensure that all the user connection through
+	// "cluster" endpoints will be admin
+	clusterAuthBackend.SetDefaultUserRole("admin")
 
-	agentWSServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/agent"))
-	_, err = NewTopologyAgentEndpoint(agentWSServer, authOptions, cached, g)
+	apiAuthBackendName := config.GetString("analyzer.auth.api.backend")
+	apiAuthBackend, err := shttp.NewAuthenticationBackendByName(apiAuthBackendName)
 	if err != nil {
 		return nil, err
 	}
 
-	publisherWSServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/publisher"))
-	_, err = NewTopologyPublisherEndpoint(publisherWSServer, authOptions, g)
+	hserver.RegisterLoginRoute(apiAuthBackend)
+
+	agentWSServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/agent", clusterAuthBackend))
+	_, err = NewTopologyAgentEndpoint(agentWSServer, cached, g)
 	if err != nil {
 		return nil, err
 	}
 
-	replicationWSServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/replication"))
-	replicationEndpoint, err := NewTopologyReplicationEndpoint(replicationWSServer, authOptions, cached, g)
+	publisherWSServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/publisher", apiAuthBackend))
+	_, err = NewTopologyPublisherEndpoint(publisherWSServer, g)
 	if err != nil {
 		return nil, err
 	}
 
 	tableClient := flow.NewTableClient(agentWSServer)
 
 	storage, err := storage.NewStorageFromConfig(etcdClient)
+
+	replicationWSServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/replication", clusterAuthBackend))
+	replicationEndpoint, err := NewTopologyReplicationEndpoint(replicationWSServer, clusterAuthOptions, cached, g)
 	if err != nil {
 		return nil, err
 	}
 
-	// declare all extension available throught API and filtering
+	// declare all extension available through API and filtering
 	tr := traversal.NewGremlinTraversalParser()
 	tr.AddTraversalExtension(ge.NewMetricsTraversalExtension())
 	tr.AddTraversalExtension(ge.NewRawPacketsTraversalExtension())
 	tr.AddTraversalExtension(ge.NewFlowTraversalExtension(tableClient, storage))
 	tr.AddTraversalExtension(ge.NewSocketsTraversalExtension())
 	tr.AddTraversalExtension(ge.NewDescendantsTraversalExtension())
 
-	subscriberWSServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/subscriber"))
+	subscriberWSServer := shttp.NewWSStructServer(shttp.NewWSServer(hserver, "/ws/subscriber", apiAuthBackend))
 	topology.NewTopologySubscriberEndpoint(subscriberWSServer, g, tr)
 
 	probeBundle, err := NewTopologyProbeBundleFromConfig(g)
 	if err != nil {
 		return nil, err
 	}
 
-	apiServer, err := api.NewAPI(hserver, etcdClient.KeysAPI, common.AnalyzerService)
+	apiServer, err := api.NewAPI(hserver, etcdClient.KeysAPI, common.AnalyzerService, apiAuthBackend)
 	if err != nil {
 		return nil, err
 	}
 
-	captureAPIHandler, err := api.RegisterCaptureAPI(apiServer, g)
+	captureAPIHandler, err := api.RegisterCaptureAPI(apiServer, g, apiAuthBackend)
 	if err != nil {
 		return nil, err
 	}
 
-	metadataAPIHandler, err := api.RegisterUserMetadataAPI(apiServer, g)
+	metadataAPIHandler, err := api.RegisterUserMetadataAPI(apiServer, g, apiAuthBackend)
 	if err != nil {
 		return nil, err
 	}
 
-	piAPIHandler, err := api.RegisterPacketInjectorAPI(g, apiServer)
+	piAPIHandler, err := api.RegisterPacketInjectorAPI(g, apiServer, apiAuthBackend)
 	if err != nil {
 		return nil, err
 	}
 	piClient := packet_injector.NewPacketInjectorClient(agentWSServer, etcdClient, piAPIHandler, g)
 
-	if _, err = api.RegisterAlertAPI(apiServer); err != nil {
+	if _, err = api.RegisterAlertAPI(apiServer, apiAuthBackend); err != nil {
 		return nil, err
 	}
 
-	if _, err := api.RegisterWorkflowAPI(apiServer); err != nil {
+	if _, err := api.RegisterWorkflowAPI(apiServer, apiAuthBackend); err != nil {
 		return nil, err
 	}
 
 	onDemandClient := ondemand.NewOnDemandProbeClient(g, captureAPIHandler, agentWSServer, subscriberWSServer, etcdClient)
 
 	metadataManager := metadata.NewUserMetadataManager(g, metadataAPIHandler)
 
-	flowServer, err := NewFlowServer(hserver, g, storage, probeBundle)
+	flowServer, err := NewFlowServer(hserver, g, storage, probeBundle, clusterAuthBackend)
 	if err != nil {
 		return nil, err
 	}
@@ -334,22 +349,34 @@ func NewServerFromConfig() (*Server, error) {
 
 	s.createStartupCapture(captureAPIHandler)
 
-	api.RegisterTopologyAPI(hserver, g, tr)
-	api.RegisterPcapAPI(hserver, storage)
-	api.RegisterConfigAPI(hserver)
-	api.RegisterStatusAPI(hserver, s)
+	api.RegisterTopologyAPI(hserver, g, tr, apiAuthBackend)
+	api.RegisterPcapAPI(hserver, storage, apiAuthBackend)
+	api.RegisterConfigAPI(hserver, apiAuthBackend)
+	api.RegisterStatusAPI(hserver, s, apiAuthBackend)
 
-	if err := dede.RegisterHandler("terminal", "/dede", hserver.Router); err != nil {
-		return nil, err
+	if config.GetBool("analyzer.ssh_enabled") {
+		if err := dede.RegisterHandler("terminal", "/dede", hserver.Router); err != nil {
+			return nil, err
+		}
+	}
+
+	// server index for the following url as the client side will redirect
+	// the user to the correct page
+	routes := []shttp.Route{
+		{Path: "/topology", Method: "GET", HandlerFunc: hserver.ServeIndex},
+		{Path: "/conversation", Method: "GET", HandlerFunc: hserver.ServeIndex},
+		{Path: "/discovery", Method: "GET", HandlerFunc: hserver.ServeIndex},
+		{Path: "/preference", Method: "GET", HandlerFunc: hserver.ServeIndex},
+		{Path: "/status", Method: "GET", HandlerFunc: hserver.ServeIndex},
 	}
+	hserver.RegisterRoutes(routes, shttp.NewNoAuthenticationBackend())
 
 	return s, nil
 }
 
-// NewAnalyzerAuthenticationOpts returns an object to authenticate to the analyzer
-func NewAnalyzerAuthenticationOpts() *shttp.AuthenticationOpts {
+func AnalyzerClusterAuthenticationOpts() *shttp.AuthenticationOpts {
 	return &shttp.AuthenticationOpts{
-		Username: config.GetString("auth.analyzer_username"),
-		Password: config.GetString("auth.analyzer_password"),
+		Username: config.GetString("analyzer.auth.cluster.username"),
+		Password: config.GetString("analyzer.auth.cluster.password"),
 	}
 }