Skip to content

Commit

Permalink
Merge pull request #773 from lpabon/sdk-auth-4
Browse files Browse the repository at this point in the history
Authentication, Authorization, HTTPS support
  • Loading branch information
lpabon authored Dec 7, 2018
2 parents 24fc8fc + b121d99 commit 7ad3786
Show file tree
Hide file tree
Showing 56 changed files with 3,887 additions and 245 deletions.
4 changes: 2 additions & 2 deletions api/server/sdk/alerts.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ const (
// alertsServer pointer properly instantiated with a valid
// alerts.filterDeleter.
type alertsServer struct {
server *Server
server serverAccessor
}

func (s *alertsServer) alert() alerts.FilterDeleter {
Expand All @@ -48,7 +48,7 @@ func (s *alertsServer) alert() alerts.FilterDeleter {
// NewAlertsServer provides an instance of alerts server interface.
func NewAlertsServer(filterDeleter alerts.FilterDeleter) api.OpenStorageAlertsServer {
return &alertsServer{
server: &Server{alertHandler: filterDeleter},
server: &sdkGrpcServer{alertHandler: filterDeleter},
}
}

Expand Down
2 changes: 1 addition & 1 deletion api/server/sdk/cloud_backup.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import (

// CloudBackupServer is an implementation of the gRPC OpenStorageCloudBackup interface
type CloudBackupServer struct {
server *Server
server serverAccessor
}

func (s *CloudBackupServer) driver() volume.VolumeDriver {
Expand Down
2 changes: 1 addition & 1 deletion api/server/sdk/cluster.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import (

// ClusterServer is an implementation of the gRPC OpenStorageClusterServer interface
type ClusterServer struct {
server *Server
server serverAccessor
}

func (s *ClusterServer) cluster() cluster.Cluster {
Expand Down
20 changes: 19 additions & 1 deletion api/server/sdk/cluster_pair.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import (

// ClusterPairServer is an implementation of the gRPC OpenStorageClusterServer interface
type ClusterPairServer struct {
server *Server
server serverAccessor
}

func (s *ClusterPairServer) cluster() cluster.Cluster {
Expand All @@ -39,6 +39,9 @@ func (s *ClusterPairServer) Create(
ctx context.Context,
req *api.SdkClusterPairCreateRequest,
) (*api.SdkClusterPairCreateResponse, error) {
if s.cluster() == nil {
return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
}

if req.GetRequest() == nil {
return nil, status.Errorf(codes.InvalidArgument, "Must supply valid request")
Expand All @@ -60,6 +63,9 @@ func (s *ClusterPairServer) Inspect(
ctx context.Context,
req *api.SdkClusterPairInspectRequest,
) (*api.SdkClusterPairInspectResponse, error) {
if s.cluster() == nil {
return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
}

if len(req.GetId()) == 0 {
return nil, status.Error(codes.InvalidArgument, "Must supply cluster ID")
Expand All @@ -78,6 +84,9 @@ func (s *ClusterPairServer) Enumerate(
ctx context.Context,
req *api.SdkClusterPairEnumerateRequest,
) (*api.SdkClusterPairEnumerateResponse, error) {
if s.cluster() == nil {
return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
}

resp, err := s.cluster().EnumeratePairs()
if err != nil {
Expand All @@ -93,6 +102,9 @@ func (s *ClusterPairServer) GetToken(
ctx context.Context,
req *api.SdkClusterPairGetTokenRequest,
) (*api.SdkClusterPairGetTokenResponse, error) {
if s.cluster() == nil {
return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
}

resp, err := s.cluster().GetPairToken(false)
if err != nil {
Expand All @@ -108,6 +120,9 @@ func (s *ClusterPairServer) ResetToken(
ctx context.Context,
req *api.SdkClusterPairResetTokenRequest,
) (*api.SdkClusterPairResetTokenResponse, error) {
if s.cluster() == nil {
return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
}

resp, err := s.cluster().GetPairToken(true)
if err != nil {
Expand All @@ -123,6 +138,9 @@ func (s *ClusterPairServer) Delete(
ctx context.Context,
req *api.SdkClusterPairDeleteRequest,
) (*api.SdkClusterPairDeleteResponse, error) {
if s.cluster() == nil {
return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
}

if len(req.GetClusterId()) == 0 {
return nil, status.Error(codes.InvalidArgument, "Must supply valid cluster ID")
Expand Down
27 changes: 17 additions & 10 deletions api/server/sdk/cluster_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package sdk

import (
"context"
"io/ioutil"
"testing"

"github.com/golang/mock/gomock"
Expand All @@ -33,27 +34,30 @@ func TestNewSdkServerBadParameters(t *testing.T) {
s, err := New(nil)
assert.Nil(t, s)
assert.NotNil(t, err)
assert.Contains(t, err.Error(), "configuration")

s, err = New(&ServerConfig{})
assert.Nil(t, s)
assert.NotNil(t, err)
assert.Contains(t, err.Error(), "Unable to setup server")
assert.Contains(t, err.Error(), "Must provide unix domain")

s, err = New(&ServerConfig{
Net: "test",
})
assert.Nil(t, s)
assert.NotNil(t, err)
assert.Contains(t, err.Error(), "Unable to setup server")
assert.Contains(t, err.Error(), "Must provide unix domain")

s, err = New(&ServerConfig{
Net: "test",
Address: "blah",
DriverName: "name",
Net: "test",
Socket: "blah",
RestPort: testRESTPort,
AccessOutput: ioutil.Discard,
AuditOutput: ioutil.Discard,
})
assert.Nil(t, s)
assert.NotNil(t, err)
assert.Contains(t, err.Error(), "Unable to get driver")
assert.Contains(t, err.Error(), "Address must be")

// Add driver to registry
mc := gomock.NewController(t)
Expand All @@ -64,13 +68,16 @@ func TestNewSdkServerBadParameters(t *testing.T) {
})
defer volumedrivers.Remove("mock")
s, err = New(&ServerConfig{
Net: "test",
Address: "blah",
DriverName: "mock",
Net: "test",
Address: "blah",
DriverName: "mock",
RestPort: testRESTPort,
AccessOutput: ioutil.Discard,
AuditOutput: ioutil.Discard,
})
assert.Nil(t, s)
assert.NotNil(t, err)
assert.Contains(t, err.Error(), "Unable to setup server")
assert.Contains(t, err.Error(), "Must provide unix domain")
}

func TestSdkClusterInspectCurrent(t *testing.T) {
Expand Down
2 changes: 1 addition & 1 deletion api/server/sdk/credentials.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import (

// CredentialServer is an implementation of the gRPC OpenStorageCredential interface
type CredentialServer struct {
server *Server
server serverAccessor
}

func (s *CredentialServer) driver() volume.VolumeDriver {
Expand Down
2 changes: 1 addition & 1 deletion api/server/sdk/identity.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import (

// IdentityServer is an implementation of the gRPC OpenStorageIdentityServer interface
type IdentityServer struct {
server *Server
server serverAccessor
}

func (s *IdentityServer) driver() volume.VolumeDriver {
Expand Down
2 changes: 1 addition & 1 deletion api/server/sdk/node.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import (

// NodeServer is an implementation of the gRPC OpenStorageNodeServer interface
type NodeServer struct {
server *Server
server serverAccessor
}

func (s *NodeServer) cluster() cluster.Cluster {
Expand Down
2 changes: 1 addition & 1 deletion api/server/sdk/objectstore.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import (

// Objectstoreserver is an implementation of the gRPC OpenStorageObjectstore interface
type ObjectstoreServer struct {
server *Server
server serverAccessor
}

func (s *ObjectstoreServer) cluster() cluster.Cluster {
Expand Down
153 changes: 153 additions & 0 deletions api/server/sdk/rest_gateway.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,153 @@
/*
Package sdk is the gRPC implementation of the SDK gRPC server
Copyright 2018 Portworx
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package sdk

import (
"context"
"fmt"
"mime"
"net/http"

"github.com/gobuffalo/packr"
"github.com/grpc-ecosystem/grpc-gateway/runtime"
"github.com/sirupsen/logrus"
"google.golang.org/grpc"

"github.com/libopenstorage/openstorage/api"
"github.com/libopenstorage/openstorage/pkg/grpcserver"
)

type sdkRestGateway struct {
config ServerConfig
restPort string
grpcServer *sdkGrpcServer
server *http.Server
}

func newSdkRestGateway(config *ServerConfig, grpcServer *sdkGrpcServer) (*sdkRestGateway, error) {
return &sdkRestGateway{
config: *config,
restPort: config.RestPort,
grpcServer: grpcServer,
}, nil
}

func (s *sdkRestGateway) Start() error {
mux, err := s.restServerSetupHandlers()
if err != nil {
return err
}

// Create object here so that we can access its Close receiver.
address := ":" + s.restPort
s.server = &http.Server{
Addr: address,
Handler: mux,
}

ready := make(chan bool)
go func() {
ready <- true
var err error
if s.config.Tls != nil {
err = s.server.ListenAndServeTLS(s.config.Tls.CertFile, s.config.Tls.KeyFile)
} else {
err = s.server.ListenAndServe()
}

if err == http.ErrServerClosed || err == nil {
return
} else {
logrus.Fatalf("Unable to start SDK REST gRPC Gateway: %s\n",
err.Error())
}
}()
<-ready
logrus.Infof("SDK gRPC REST Gateway started on port :%s", s.restPort)

return nil
}

func (s *sdkRestGateway) Stop() {
if err := s.server.Close(); err != nil {
logrus.Fatalf("REST GW STOP error: %v", err)
}
}

// restServerSetupHandlers sets up the handlers to the swagger ui and
// to the gRPC REST Gateway.
func (s *sdkRestGateway) restServerSetupHandlers() (*http.ServeMux, error) {

// Create an HTTP server router
mux := http.NewServeMux()

// Swagger files using packr
swaggerUIBox := packr.NewBox("./swagger-ui")
swaggerJSONBox := packr.NewBox("./api")
mime.AddExtensionType(".svg", "image/svg+xml")

// Handler to return swagger.json
mux.HandleFunc("/swagger.json", func(w http.ResponseWriter, r *http.Request) {
w.Write(swaggerJSONBox.Bytes("api.swagger.json"))
})

// Handler to access the swagger ui. The UI pulls the swagger
// json file from /swagger.json
// The link below MUST have th last '/'. It is really important.
prefix := "/swagger-ui/"
mux.Handle(prefix,
http.StripPrefix(prefix, http.FileServer(swaggerUIBox)))

// Create a router just for HTTP REST gRPC Server Gateway
gmux := runtime.NewServeMux()

// Connect to gRPC unix domain socket
conn, err := grpcserver.Connect(
s.grpcServer.Address(),
[]grpc.DialOption{grpc.WithInsecure()})
if err != nil {
return nil, fmt.Errorf("Failed to connect to gRPC handler: %v", err)
}

// REST Gateway Handlers
handlers := []func(context.Context, *runtime.ServeMux, *grpc.ClientConn) (err error){
api.RegisterOpenStorageClusterHandler,
api.RegisterOpenStorageNodeHandler,
api.RegisterOpenStorageVolumeHandler,
api.RegisterOpenStorageObjectstoreHandler,
api.RegisterOpenStorageCredentialsHandler,
api.RegisterOpenStorageSchedulePolicyHandler,
api.RegisterOpenStorageCloudBackupHandler,
api.RegisterOpenStorageIdentityHandler,
api.RegisterOpenStorageMountAttachHandler,
api.RegisterOpenStorageAlertsHandler,
api.RegisterOpenStorageClusterPairHandler,
api.RegisterOpenStorageMigrateHandler,
}

// Register the REST Gateway handlers
for _, handler := range handlers {
err := handler(context.Background(), gmux, conn)
if err != nil {
return nil, err
}
}

// Pass all other unhandled paths to the gRPC gateway
mux.Handle("/", gmux)
return mux, nil
}
2 changes: 1 addition & 1 deletion api/server/sdk/schedule_policy.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import (

// SchedulePolicyServer is an implementation of the gRPC OpenStorageSchedulePolicy interface
type SchedulePolicyServer struct {
server *Server
server serverAccessor
}

func (s *SchedulePolicyServer) cluster() cluster.Cluster {
Expand Down
Loading

0 comments on commit 7ad3786

Please sign in to comment.