fluidB
/
fluidb-old
1
0
Fork 0
Code Pull Requests Activity

Merge pull request #1057 from ngharrington/nharrington-forward-creds

Browse Source 
WIP TCP service auth
master
Philip O'Toole 2 years ago committed by GitHub
parent 8873141904 f750f78558
commit 7211e4d93b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 658 additions and 175 deletions
Show Stats Download Patch File Download Diff File

9
auth/credential_store_test.go
Unescape Escape View File

@ -227,10 +227,19 @@ func Test_AuthPermsAA(t *testing.T) {
if store.AA("nonexistent", "password1", "foo") {
t.Fatalf("nonexistent authenticated and authorized for foo")
}
if !store.AA("nonexistent", "password1", "qux") {
t.Fatalf("nonexistent not authenticated and authorized for qux")
}
// explicit check of anonymous user
if !store.AA("", "", "qux") {
t.Fatalf("anonymous incorrectly not authorized")
}
if store.AA("", "", "foo") {
t.Fatalf("anonymous incorrectly authorized")
}
if !store.AA("username1", "password1", "foo") {
t.Fatalf("username1 not authenticated and authorized for foo")
}

31
cluster/client.go
Unescape Escape View File

@ -132,8 +132,10 @@ func (c *Client) GetNodeAPIAddr(nodeAddr string, timeout time.Duration) (string,
return a.Url, nil
}
// Execute performs an Execute on a remote node.
func (c *Client) Execute(er *command.ExecuteRequest, nodeAddr string, timeout time.Duration) ([]*command.ExecuteResult, error) {
// Execute performs an Execute on a remote node. If username is an empty string
// no credential information will be included in the Execute request to the
// remote node.
func (c *Client) Execute(er *command.ExecuteRequest, nodeAddr string, username string, password string, timeout time.Duration) ([]*command.ExecuteResult, error) {
conn, err := c.dial(nodeAddr, c.timeout)
if err != nil {
return nil, err
@ -147,6 +149,14 @@ func (c *Client) Execute(er *command.ExecuteRequest, nodeAddr string, timeout ti
ExecuteRequest: er,
},
}
if username != "" {
command.Credentials = &Credentials{
Username: username,
Password: password,
}
}
p, err := proto.Marshal(command)
if err != nil {
return nil, fmt.Errorf("command marshal: %s", err)
@ -205,14 +215,14 @@ func (c *Client) Execute(er *command.ExecuteRequest, nodeAddr string, timeout ti
return nil, err
}
if a.Error != "" {
return nil, errors.New(a.Error)
if a.Error != nil {
return nil, errors.New(a.Error.GetError())
}
return a.Results, nil
}
// Query performs a Query on a remote node.
func (c *Client) Query(qr *command.QueryRequest, nodeAddr string, timeout time.Duration) ([]*command.QueryRows, error) {
func (c *Client) Query(qr *command.QueryRequest, nodeAddr string, username string, password string, timeout time.Duration) ([]*command.QueryRows, error) {
conn, err := c.dial(nodeAddr, c.timeout)
if err != nil {
return nil, err
@ -226,6 +236,13 @@ func (c *Client) Query(qr *command.QueryRequest, nodeAddr string, timeout time.D
QueryRequest: qr,
},
}
if username != "" {
command.Credentials = &Credentials{
Username: username,
Password: password,
}
}
p, err := proto.Marshal(command)
if err != nil {
return nil, fmt.Errorf("command marshal: %s", err)
@ -281,8 +298,8 @@ func (c *Client) Query(qr *command.QueryRequest, nodeAddr string, timeout time.D
return nil, err
}
if a.Error != "" {
return nil, errors.New(a.Error)
if a.Error != nil {
return nil, errors.New(a.Error.GetError())
}
return a.Rows, nil
}

452
cluster/message.pb.go
Unescape Escape View File

@ -1,8 +1,8 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.27.1
// protoc v3.13.0
// source: message.proto
// protoc-gen-go v1.28.0
// protoc v3.21.2
// source: cluster/message.proto
package cluster
@ -57,11 +57,11 @@ func (x Command_Type) String() string {
}
func (Command_Type) Descriptor() protoreflect.EnumDescriptor {
return file_message_proto_enumTypes[0].Descriptor()
return file_cluster_message_proto_enumTypes[0].Descriptor()
}
func (Command_Type) Type() protoreflect.EnumType {
return &file_message_proto_enumTypes[0]
return &file_cluster_message_proto_enumTypes[0]
}
func (x Command_Type) Number() protoreflect.EnumNumber {
@ -70,7 +70,111 @@ func (x Command_Type) Number() protoreflect.EnumNumber {
// Deprecated: Use Command_Type.Descriptor instead.
func (Command_Type) EnumDescriptor() ([]byte, []int) {
return file_message_proto_rawDescGZIP(), []int{1, 0}
return file_cluster_message_proto_rawDescGZIP(), []int{2, 0}
}
type ResultsError_Type int32
const (
ResultsError_RESULTS_ERROR_UNSPECIFIED ResultsError_Type = 0
ResultsError_RESULTS_ERROR_UNAUTHORIZED ResultsError_Type = 1
ResultsError_RESULTS_ERROR_NIL_QUERY_REQUEST ResultsError_Type = 2
)
// Enum value maps for ResultsError_Type.
var (
ResultsError_Type_name = map[int32]string{
0: "RESULTS_ERROR_UNSPECIFIED",
1: "RESULTS_ERROR_UNAUTHORIZED",
2: "RESULTS_ERROR_NIL_QUERY_REQUEST",
}
ResultsError_Type_value = map[string]int32{
"RESULTS_ERROR_UNSPECIFIED": 0,
"RESULTS_ERROR_UNAUTHORIZED": 1,
"RESULTS_ERROR_NIL_QUERY_REQUEST": 2,
}
)
func (x ResultsError_Type) Enum() *ResultsError_Type {
p := new(ResultsError_Type)
*p = x
return p
}
func (x ResultsError_Type) String() string {
return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
}
func (ResultsError_Type) Descriptor() protoreflect.EnumDescriptor {
return file_cluster_message_proto_enumTypes[1].Descriptor()
}
func (ResultsError_Type) Type() protoreflect.EnumType {
return &file_cluster_message_proto_enumTypes[1]
}
func (x ResultsError_Type) Number() protoreflect.EnumNumber {
return protoreflect.EnumNumber(x)
}
// Deprecated: Use ResultsError_Type.Descriptor instead.
func (ResultsError_Type) EnumDescriptor() ([]byte, []int) {
return file_cluster_message_proto_rawDescGZIP(), []int{3, 0}
}
type Credentials struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Username string `protobuf:"bytes,1,opt,name=username,proto3" json:"username,omitempty"`
Password string `protobuf:"bytes,2,opt,name=password,proto3" json:"password,omitempty"`
}
func (x *Credentials) Reset() {
*x = Credentials{}
if protoimpl.UnsafeEnabled {
mi := &file_cluster_message_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *Credentials) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Credentials) ProtoMessage() {}
func (x *Credentials) ProtoReflect() protoreflect.Message {
mi := &file_cluster_message_proto_msgTypes[0]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Credentials.ProtoReflect.Descriptor instead.
func (*Credentials) Descriptor() ([]byte, []int) {
return file_cluster_message_proto_rawDescGZIP(), []int{0}
}
func (x *Credentials) GetUsername() string {
if x != nil {
return x.Username
}
return ""
}
func (x *Credentials) GetPassword() string {
if x != nil {
return x.Password
}
return ""
}
type Address struct {
@ -84,7 +188,7 @@ type Address struct {
func (x *Address) Reset() {
*x = Address{}
if protoimpl.UnsafeEnabled {
mi := &file_message_proto_msgTypes[0]
mi := &file_cluster_message_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -97,7 +201,7 @@ func (x *Address) String() string {
func (*Address) ProtoMessage() {}
func (x *Address) ProtoReflect() protoreflect.Message {
mi := &file_message_proto_msgTypes[0]
mi := &file_cluster_message_proto_msgTypes[1]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -110,7 +214,7 @@ func (x *Address) ProtoReflect() protoreflect.Message {
// Deprecated: Use Address.ProtoReflect.Descriptor instead.
func (*Address) Descriptor() ([]byte, []int) {
return file_message_proto_rawDescGZIP(), []int{0}
return file_cluster_message_proto_rawDescGZIP(), []int{1}
}
func (x *Address) GetUrl() string {
@ -130,12 +234,13 @@ type Command struct {
// *Command_ExecuteRequest
// *Command_QueryRequest
Request isCommand_Request `protobuf_oneof:"request"`
Credentials *Credentials `protobuf:"bytes,4,opt,name=credentials,proto3" json:"credentials,omitempty"`
}
func (x *Command) Reset() {
*x = Command{}
if protoimpl.UnsafeEnabled {
mi := &file_message_proto_msgTypes[1]
mi := &file_cluster_message_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -148,7 +253,7 @@ func (x *Command) String() string {
func (*Command) ProtoMessage() {}
func (x *Command) ProtoReflect() protoreflect.Message {
mi := &file_message_proto_msgTypes[1]
mi := &file_cluster_message_proto_msgTypes[2]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -161,7 +266,7 @@ func (x *Command) ProtoReflect() protoreflect.Message {
// Deprecated: Use Command.ProtoReflect.Descriptor instead.
func (*Command) Descriptor() ([]byte, []int) {
return file_message_proto_rawDescGZIP(), []int{1}
return file_cluster_message_proto_rawDescGZIP(), []int{2}
}
func (x *Command) GetType() Command_Type {
@ -192,6 +297,13 @@ func (x *Command) GetQueryRequest() *command.QueryRequest {
return nil
}
func (x *Command) GetCredentials() *Credentials {
if x != nil {
return x.Credentials
}
return nil
}
type isCommand_Request interface {
isCommand_Request()
}
@ -208,19 +320,74 @@ func (*Command_ExecuteRequest) isCommand_Request() {}
func (*Command_QueryRequest) isCommand_Request() {}
type ResultsError struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Type ResultsError_Type `protobuf:"varint,1,opt,name=type,proto3,enum=cluster.ResultsError_Type" json:"type,omitempty"`
Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"`
}
func (x *ResultsError) Reset() {
*x = ResultsError{}
if protoimpl.UnsafeEnabled {
mi := &file_cluster_message_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *ResultsError) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ResultsError) ProtoMessage() {}
func (x *ResultsError) ProtoReflect() protoreflect.Message {
mi := &file_cluster_message_proto_msgTypes[3]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use ResultsError.ProtoReflect.Descriptor instead.
func (*ResultsError) Descriptor() ([]byte, []int) {
return file_cluster_message_proto_rawDescGZIP(), []int{3}
}
func (x *ResultsError) GetType() ResultsError_Type {
if x != nil {
return x.Type
}
return ResultsError_RESULTS_ERROR_UNSPECIFIED
}
func (x *ResultsError) GetError() string {
if x != nil {
return x.Error
}
return ""
}
type CommandExecuteResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
Error *ResultsError `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
Results []*command.ExecuteResult `protobuf:"bytes,2,rep,name=results,proto3" json:"results,omitempty"`
}
func (x *CommandExecuteResponse) Reset() {
*x = CommandExecuteResponse{}
if protoimpl.UnsafeEnabled {
mi := &file_message_proto_msgTypes[2]
mi := &file_cluster_message_proto_msgTypes[4]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -233,7 +400,7 @@ func (x *CommandExecuteResponse) String() string {
func (*CommandExecuteResponse) ProtoMessage() {}
func (x *CommandExecuteResponse) ProtoReflect() protoreflect.Message {
mi := &file_message_proto_msgTypes[2]
mi := &file_cluster_message_proto_msgTypes[4]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -246,14 +413,14 @@ func (x *CommandExecuteResponse) ProtoReflect() protoreflect.Message {
// Deprecated: Use CommandExecuteResponse.ProtoReflect.Descriptor instead.
func (*CommandExecuteResponse) Descriptor() ([]byte, []int) {
return file_message_proto_rawDescGZIP(), []int{2}
return file_cluster_message_proto_rawDescGZIP(), []int{4}
}
func (x *CommandExecuteResponse) GetError() string {
func (x *CommandExecuteResponse) GetError() *ResultsError {
if x != nil {
return x.Error
}
return ""
return nil
}
func (x *CommandExecuteResponse) GetResults() []*command.ExecuteResult {
@ -268,14 +435,14 @@ type CommandQueryResponse struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
Error *ResultsError `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
Rows []*command.QueryRows `protobuf:"bytes,2,rep,name=rows,proto3" json:"rows,omitempty"`
}
func (x *CommandQueryResponse) Reset() {
*x = CommandQueryResponse{}
if protoimpl.UnsafeEnabled {
mi := &file_message_proto_msgTypes[3]
mi := &file_cluster_message_proto_msgTypes[5]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -288,7 +455,7 @@ func (x *CommandQueryResponse) String() string {
func (*CommandQueryResponse) ProtoMessage() {}
func (x *CommandQueryResponse) ProtoReflect() protoreflect.Message {
mi := &file_message_proto_msgTypes[3]
mi := &file_cluster_message_proto_msgTypes[5]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -301,14 +468,14 @@ func (x *CommandQueryResponse) ProtoReflect() protoreflect.Message {
// Deprecated: Use CommandQueryResponse.ProtoReflect.Descriptor instead.
func (*CommandQueryResponse) Descriptor() ([]byte, []int) {
return file_message_proto_rawDescGZIP(), []int{3}
return file_cluster_message_proto_rawDescGZIP(), []int{5}
}
func (x *CommandQueryResponse) GetError() string {
func (x *CommandQueryResponse) GetError() *ResultsError {
if x != nil {
return x.Error
}
return ""
return nil
}
func (x *CommandQueryResponse) GetRows() []*command.QueryRows {
@ -318,95 +485,138 @@ func (x *CommandQueryResponse) GetRows() []*command.QueryRows {
return nil
}
var File_message_proto protoreflect.FileDescriptor
var file_message_proto_rawDesc = []byte{
0x0a, 0x0d, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12,
0x07, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x1a, 0x15, 0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e,
0x64, 0x2f, 0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22,
0x1b, 0x0a, 0x07, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x72,
0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x72, 0x6c, 0x22, 0xb8, 0x02, 0x0a,
0x07, 0x43, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x12, 0x29, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65,
0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x15, 0x2e, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72,
0x2e, 0x43, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74,
0x79, 0x70, 0x65, 0x12, 0x42, 0x0a, 0x0f, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x5f, 0x72,
0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x63,
0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x48, 0x00, 0x52, 0x0e, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65,
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3c, 0x0a, 0x0d, 0x71, 0x75, 0x65, 0x72, 0x79,
0x5f, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15,
0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e, 0x51, 0x75, 0x65, 0x72, 0x79, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x48, 0x00, 0x52, 0x0c, 0x71, 0x75, 0x65, 0x72, 0x79, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x75, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x18, 0x0a,
0x14, 0x43, 0x4f, 0x4d, 0x4d, 0x41, 0x4e, 0x44, 0x5f, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e,
0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x21, 0x0a, 0x1d, 0x43, 0x4f, 0x4d, 0x4d, 0x41,
0x4e, 0x44, 0x5f, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x47, 0x45, 0x54, 0x5f, 0x4e, 0x4f, 0x44, 0x45,
0x5f, 0x41, 0x50, 0x49, 0x5f, 0x55, 0x52, 0x4c, 0x10, 0x01, 0x12, 0x18, 0x0a, 0x14, 0x43, 0x4f,
0x4d, 0x4d, 0x41, 0x4e, 0x44, 0x5f, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x45, 0x58, 0x45, 0x43, 0x55,
0x54, 0x45, 0x10, 0x02, 0x12, 0x16, 0x0a, 0x12, 0x43, 0x4f, 0x4d, 0x4d, 0x41, 0x4e, 0x44, 0x5f,
0x54, 0x59, 0x50, 0x45, 0x5f, 0x51, 0x55, 0x45, 0x52, 0x59, 0x10, 0x03, 0x42, 0x09, 0x0a, 0x07,
0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x60, 0x0a, 0x16, 0x43, 0x6f, 0x6d, 0x6d, 0x61,
0x6e, 0x64, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x12, 0x30, 0x0a, 0x07, 0x72, 0x65, 0x73, 0x75, 0x6c,
0x74, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x61,
0x6e, 0x64, 0x2e, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74,
0x52, 0x07, 0x72, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x22, 0x54, 0x0a, 0x14, 0x43, 0x6f, 0x6d,
0x6d, 0x61, 0x6e, 0x64, 0x51, 0x75, 0x65, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x12, 0x26, 0x0a, 0x04, 0x72, 0x6f, 0x77, 0x73, 0x18,
0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e,
0x51, 0x75, 0x65, 0x72, 0x79, 0x52, 0x6f, 0x77, 0x73, 0x52, 0x04, 0x72, 0x6f, 0x77, 0x73, 0x42,
0x22, 0x5a, 0x20, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x72, 0x71,
0x6c, 0x69, 0x74, 0x65, 0x2f, 0x72, 0x71, 0x6c, 0x69, 0x74, 0x65, 0x2f, 0x63, 0x6c, 0x75, 0x73,
0x74, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
var File_cluster_message_proto protoreflect.FileDescriptor
var file_cluster_message_proto_rawDesc = []byte{
0x0a, 0x15, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x2f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67,
0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x07, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72,
0x1a, 0x15, 0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2f, 0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e,
0x64, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x45, 0x0a, 0x0b, 0x43, 0x72, 0x65, 0x64, 0x65,
0x6e, 0x74, 0x69, 0x61, 0x6c, 0x73, 0x12, 0x1a, 0x0a, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61,
0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61,
0x6d, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x18, 0x02,
0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x22, 0x1b,
0x0a, 0x07, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x72, 0x6c,
0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x72, 0x6c, 0x22, 0xf0, 0x02, 0x0a, 0x07,
0x43, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x12, 0x29, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18,
0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x15, 0x2e, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x2e,
0x43, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79,
0x70, 0x65, 0x12, 0x42, 0x0a, 0x0f, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x5f, 0x72, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x63, 0x6f,
0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65, 0x71,
0x75, 0x65, 0x73, 0x74, 0x48, 0x00, 0x52, 0x0e, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52,
0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3c, 0x0a, 0x0d, 0x71, 0x75, 0x65, 0x72, 0x79, 0x5f,
0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e,
0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e, 0x51, 0x75, 0x65, 0x72, 0x79, 0x52, 0x65, 0x71,
0x75, 0x65, 0x73, 0x74, 0x48, 0x00, 0x52, 0x0c, 0x71, 0x75, 0x65, 0x72, 0x79, 0x52, 0x65, 0x71,
0x75, 0x65, 0x73, 0x74, 0x12, 0x36, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69,
0x61, 0x6c, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x63, 0x6c, 0x75, 0x73,
0x74, 0x65, 0x72, 0x2e, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x73, 0x52,
0x0b, 0x63, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x73, 0x22, 0x75, 0x0a, 0x04,
0x54, 0x79, 0x70, 0x65, 0x12, 0x18, 0x0a, 0x14, 0x43, 0x4f, 0x4d, 0x4d, 0x41, 0x4e, 0x44, 0x5f,
0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x21,
0x0a, 0x1d, 0x43, 0x4f, 0x4d, 0x4d, 0x41, 0x4e, 0x44, 0x5f, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x47,
0x45, 0x54, 0x5f, 0x4e, 0x4f, 0x44, 0x45, 0x5f, 0x41, 0x50, 0x49, 0x5f, 0x55, 0x52, 0x4c, 0x10,
0x01, 0x12, 0x18, 0x0a, 0x14, 0x43, 0x4f, 0x4d, 0x4d, 0x41, 0x4e, 0x44, 0x5f, 0x54, 0x59, 0x50,
0x45, 0x5f, 0x45, 0x58, 0x45, 0x43, 0x55, 0x54, 0x45, 0x10, 0x02, 0x12, 0x16, 0x0a, 0x12, 0x43,
0x4f, 0x4d, 0x4d, 0x41, 0x4e, 0x44, 0x5f, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x51, 0x55, 0x45, 0x52,
0x59, 0x10, 0x03, 0x42, 0x09, 0x0a, 0x07, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0xc0,
0x01, 0x0a, 0x0c, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x12,
0x2e, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x1a, 0x2e,
0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x45,
0x72, 0x72, 0x6f, 0x72, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12,
0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05,
0x65, 0x72, 0x72, 0x6f, 0x72, 0x22, 0x6a, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1d, 0x0a,
0x19, 0x52, 0x45, 0x53, 0x55, 0x4c, 0x54, 0x53, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x5f, 0x55,
0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1e, 0x0a, 0x1a,
0x52, 0x45, 0x53, 0x55, 0x4c, 0x54, 0x53, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x5f, 0x55, 0x4e,
0x41, 0x55, 0x54, 0x48, 0x4f, 0x52, 0x49, 0x5a, 0x45, 0x44, 0x10, 0x01, 0x12, 0x23, 0x0a, 0x1f,
0x52, 0x45, 0x53, 0x55, 0x4c, 0x54, 0x53, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x5f, 0x4e, 0x49,
0x4c, 0x5f, 0x51, 0x55, 0x45, 0x52, 0x59, 0x5f, 0x52, 0x45, 0x51, 0x55, 0x45, 0x53, 0x54, 0x10,
0x02, 0x22, 0x77, 0x0a, 0x16, 0x43, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x45, 0x78, 0x65, 0x63,
0x75, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x2b, 0x0a, 0x05, 0x65,
0x72, 0x72, 0x6f, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x63, 0x6c, 0x75,
0x73, 0x74, 0x65, 0x72, 0x2e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x45, 0x72, 0x72, 0x6f,
0x72, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x12, 0x30, 0x0a, 0x07, 0x72, 0x65, 0x73, 0x75,
0x6c, 0x74, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x63, 0x6f, 0x6d, 0x6d,
0x61, 0x6e, 0x64, 0x2e, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6c,
0x74, 0x52, 0x07, 0x72, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x73, 0x22, 0x6b, 0x0a, 0x14, 0x43, 0x6f,
0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x51, 0x75, 0x65, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
0x73, 0x65, 0x12, 0x2b, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28,
0x0b, 0x32, 0x15, 0x2e, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x52, 0x65, 0x73, 0x75,
0x6c, 0x74, 0x73, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x12,
0x26, 0x0a, 0x04, 0x72, 0x6f, 0x77, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e,
0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x2e, 0x51, 0x75, 0x65, 0x72, 0x79, 0x52, 0x6f, 0x77,
0x73, 0x52, 0x04, 0x72, 0x6f, 0x77, 0x73, 0x42, 0x22, 0x5a, 0x20, 0x67, 0x69, 0x74, 0x68, 0x75,
0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x72, 0x71, 0x6c, 0x69, 0x74, 0x65, 0x2f, 0x72, 0x71, 0x6c,
0x69, 0x74, 0x65, 0x2f, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f,
0x74, 0x6f, 0x33,
}
var (
file_message_proto_rawDescOnce sync.Once
file_message_proto_rawDescData = file_message_proto_rawDesc
file_cluster_message_proto_rawDescOnce sync.Once
file_cluster_message_proto_rawDescData = file_cluster_message_proto_rawDesc
)
func file_message_proto_rawDescGZIP() []byte {
file_message_proto_rawDescOnce.Do(func() {
file_message_proto_rawDescData = protoimpl.X.CompressGZIP(file_message_proto_rawDescData)
func file_cluster_message_proto_rawDescGZIP() []byte {
file_cluster_message_proto_rawDescOnce.Do(func() {
file_cluster_message_proto_rawDescData = protoimpl.X.CompressGZIP(file_cluster_message_proto_rawDescData)
})
return file_message_proto_rawDescData
return file_cluster_message_proto_rawDescData
}
var file_message_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
var file_message_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
var file_message_proto_goTypes = []interface{}{
var file_cluster_message_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
var file_cluster_message_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
var file_cluster_message_proto_goTypes = []interface{}{
(Command_Type)(0), // 0: cluster.Command.Type
(*Address)(nil), // 1: cluster.Address
(*Command)(nil), // 2: cluster.Command
(*CommandExecuteResponse)(nil), // 3: cluster.CommandExecuteResponse
(*CommandQueryResponse)(nil), // 4: cluster.CommandQueryResponse
(*command.ExecuteRequest)(nil), // 5: command.ExecuteRequest
(*command.QueryRequest)(nil), // 6: command.QueryRequest
(*command.ExecuteResult)(nil), // 7: command.ExecuteResult
(*command.QueryRows)(nil), // 8: command.QueryRows
}
var file_message_proto_depIdxs = []int32{
(ResultsError_Type)(0), // 1: cluster.ResultsError.Type
(*Credentials)(nil), // 2: cluster.Credentials
(*Address)(nil), // 3: cluster.Address
(*Command)(nil), // 4: cluster.Command
(*ResultsError)(nil), // 5: cluster.ResultsError
(*CommandExecuteResponse)(nil), // 6: cluster.CommandExecuteResponse
(*CommandQueryResponse)(nil), // 7: cluster.CommandQueryResponse
(*command.ExecuteRequest)(nil), // 8: command.ExecuteRequest
(*command.QueryRequest)(nil), // 9: command.QueryRequest
(*command.ExecuteResult)(nil), // 10: command.ExecuteResult
(*command.QueryRows)(nil), // 11: command.QueryRows
}
var file_cluster_message_proto_depIdxs = []int32{
0, // 0: cluster.Command.type:type_name -> cluster.Command.Type
5, // 1: cluster.Command.execute_request:type_name -> command.ExecuteRequest
6, // 2: cluster.Command.query_request:type_name -> command.QueryRequest
7, // 3: cluster.CommandExecuteResponse.results:type_name -> command.ExecuteResult
8, // 4: cluster.CommandQueryResponse.rows:type_name -> command.QueryRows
5, // [5:5] is the sub-list for method output_type
5, // [5:5] is the sub-list for method input_type
5, // [5:5] is the sub-list for extension type_name
5, // [5:5] is the sub-list for extension extendee
0, // [0:5] is the sub-list for field type_name
}
func init() { file_message_proto_init() }
func file_message_proto_init() {
if File_message_proto != nil {
8, // 1: cluster.Command.execute_request:type_name -> command.ExecuteRequest
9, // 2: cluster.Command.query_request:type_name -> command.QueryRequest
2, // 3: cluster.Command.credentials:type_name -> cluster.Credentials
1, // 4: cluster.ResultsError.type:type_name -> cluster.ResultsError.Type
5, // 5: cluster.CommandExecuteResponse.error:type_name -> cluster.ResultsError
10, // 6: cluster.CommandExecuteResponse.results:type_name -> command.ExecuteResult
5, // 7: cluster.CommandQueryResponse.error:type_name -> cluster.ResultsError
11, // 8: cluster.CommandQueryResponse.rows:type_name -> command.QueryRows
9, // [9:9] is the sub-list for method output_type
9, // [9:9] is the sub-list for method input_type
9, // [9:9] is the sub-list for extension type_name
9, // [9:9] is the sub-list for extension extendee
0, // [0:9] is the sub-list for field type_name
}
func init() { file_cluster_message_proto_init() }
func file_cluster_message_proto_init() {
if File_cluster_message_proto != nil {
return
}
if !protoimpl.UnsafeEnabled {
file_message_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
file_cluster_message_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Credentials); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_cluster_message_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Address); i {
case 0:
return &v.state
@ -418,7 +628,7 @@ func file_message_proto_init() {
return nil
}
}
file_message_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
file_cluster_message_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Command); i {
case 0:
return &v.state
@ -430,7 +640,19 @@ func file_message_proto_init() {
return nil
}
}
file_message_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
file_cluster_message_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*ResultsError); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_cluster_message_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*CommandExecuteResponse); i {
case 0:
return &v.state
@ -442,7 +664,7 @@ func file_message_proto_init() {
return nil
}
}
file_message_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
file_cluster_message_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*CommandQueryResponse); i {
case 0:
return &v.state
@ -455,7 +677,7 @@ func file_message_proto_init() {
}
}
}
file_message_proto_msgTypes[1].OneofWrappers = []interface{}{
file_cluster_message_proto_msgTypes[2].OneofWrappers = []interface{}{
(*Command_ExecuteRequest)(nil),
(*Command_QueryRequest)(nil),
}
@ -463,19 +685,19 @@ func file_message_proto_init() {
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_message_proto_rawDesc,
NumEnums: 1,
NumMessages: 4,
RawDescriptor: file_cluster_message_proto_rawDesc,
NumEnums: 2,
NumMessages: 6,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_message_proto_goTypes,
DependencyIndexes: file_message_proto_depIdxs,
EnumInfos: file_message_proto_enumTypes,
MessageInfos: file_message_proto_msgTypes,
GoTypes: file_cluster_message_proto_goTypes,
DependencyIndexes: file_cluster_message_proto_depIdxs,
EnumInfos: file_cluster_message_proto_enumTypes,
MessageInfos: file_cluster_message_proto_msgTypes,
}.Build()
File_message_proto = out.File
file_message_proto_rawDesc = nil
file_message_proto_goTypes = nil
file_message_proto_depIdxs = nil
File_cluster_message_proto = out.File
file_cluster_message_proto_rawDesc = nil
file_cluster_message_proto_goTypes = nil
file_cluster_message_proto_depIdxs = nil
}

21
cluster/message.proto
Unescape Escape View File

@ -5,6 +5,11 @@ import "command/command.proto";
option go_package = "github.com/rqlite/rqlite/cluster";
message Credentials {
string username = 1;
string password = 2;
}
message Address {
string url = 1;
}
@ -22,14 +27,26 @@ message Command {
command.ExecuteRequest execute_request = 2;
command.QueryRequest query_request = 3;
}
Credentials credentials = 4;
}
message ResultsError {
enum Type {
RESULTS_ERROR_UNSPECIFIED = 0;
RESULTS_ERROR_UNAUTHORIZED = 1;
RESULTS_ERROR_NIL_QUERY_REQUEST =2;
}
Type type = 1;
string error = 2;
}
message CommandExecuteResponse {
string error = 1;
ResultsError error = 1;
repeated command.ExecuteResult results = 2;
}
message CommandQueryResponse {
string error = 1;
ResultsError error = 1;
repeated command.QueryRows rows = 2;
}

57
cluster/service.go
Unescape Escape View File

@ -12,6 +12,7 @@ import (
"sync"
"time"
"github.com/rqlite/rqlite/auth"
"github.com/rqlite/rqlite/command"
"google.golang.org/protobuf/proto"
)
@ -63,6 +64,12 @@ type Database interface {
Query(qr *command.QueryRequest) ([]*command.QueryRows, error)
}
// CredentialStore is the interface credential stores must support.
type CredentialStore interface {
// AA authenticates and checks authorization for the given perm.
AA(username, password, perm string) bool
}
// Transport is the interface the network layer must provide.
type Transport interface {
net.Listener
@ -76,6 +83,8 @@ type Service struct {
db Database // The queryable system.
credentialStore CredentialStore
mu sync.RWMutex
https bool // Serving HTTPS?
apiAddr string // host:port this node serves the HTTP API.
@ -84,12 +93,13 @@ type Service struct {
}
// New returns a new instance of the cluster service
func New(tn Transport, db Database) *Service {
func New(tn Transport, db Database, credentialStore CredentialStore) *Service {
return &Service{
tn: tn,
addr: tn.Addr(),
db: db,
logger: log.New(os.Stderr, "[cluster] ", log.LstdFlags),
credentialStore: credentialStore,
}
}
@ -167,6 +177,20 @@ func (s *Service) serve() error {
}
}
func (s *Service) checkCommandPerm(c *Command, perm string) bool {
if s.credentialStore == nil {
return true
}
username := ""
password := ""
if c.Credentials != nil {
username = c.Credentials.GetUsername()
password = c.Credentials.GetPassword()
}
return s.credentialStore.AA(username, password, perm)
}
func (s *Service) handleConn(conn net.Conn) {
defer conn.Close()
@ -211,14 +235,24 @@ func (s *Service) handleConn(conn net.Conn) {
stats.Add(numExecuteRequest, 1)
resp := &CommandExecuteResponse{}
er := c.GetExecuteRequest()
if er == nil {
resp.Error = "ExecuteRequest is nil"
resp.Error = &ResultsError{
Type: ResultsError_RESULTS_ERROR_NIL_QUERY_REQUEST,
Error: "Query Request is nil",
}
} else if !s.checkCommandPerm(c, auth.PermExecute) {
resp.Error = &ResultsError{
Type: ResultsError_RESULTS_ERROR_UNAUTHORIZED,
Error: "Unauthorized",
}
} else {
res, err := s.db.Execute(er)
if err != nil {
resp.Error = err.Error()
resp.Error = &ResultsError{
Type: ResultsError_RESULTS_ERROR_UNSPECIFIED,
Error: err.Error(),
}
} else {
resp.Results = make([]*command.ExecuteResult, len(res))
for i := range res {
@ -244,11 +278,22 @@ func (s *Service) handleConn(conn net.Conn) {
qr := c.GetQueryRequest()
if qr == nil {
resp.Error = "QueryRequest is nil"
resp.Error = &ResultsError{
Type: ResultsError_RESULTS_ERROR_NIL_QUERY_REQUEST,
Error: "Query Request is nil",
}
} else if !s.checkCommandPerm(c, auth.PermQuery) {
resp.Error = &ResultsError{
Type: ResultsError_RESULTS_ERROR_UNAUTHORIZED,
Error: "Unauthorized",
}
} else {
res, err := s.db.Query(qr)
if err != nil {
resp.Error = err.Error()
resp.Error = &ResultsError{
Type: ResultsError_RESULTS_ERROR_UNSPECIFIED,
Error: err.Error(),
}
} else {
resp.Rows = make([]*command.QueryRows, len(res))
for i := range res {

32
cluster/service_database_test.go
Unescape Escape View File

@ -16,12 +16,18 @@ import (
const shortWait = 1 * time.Second
const longWait = 5 * time.Second
const (
NO_USERNAME = ""
NO_PASSWORD = ""
)
func Test_ServiceExecute(t *testing.T) {
ln, mux := mustNewMux()
go mux.Serve()
tn := mux.Listen(1) // Could be any byte value.
db := mustNewMockDatabase()
s := New(tn, db)
cred := mustNewMockCredentialStore()
s := New(tn, db, cred)
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -39,7 +45,7 @@ func Test_ServiceExecute(t *testing.T) {
}
return nil, errors.New("execute failed")
}
_, err := c.Execute(executeRequestFromString("some SQL"), s.Addr(), longWait)
_, err := c.Execute(executeRequestFromString("some SQL"), s.Addr(), NO_USERNAME, NO_PASSWORD, longWait)
if err == nil {
t.Fatalf("client failed to report error")
}
@ -57,7 +63,7 @@ func Test_ServiceExecute(t *testing.T) {
}
return []*command.ExecuteResult{result}, nil
}
res, err := c.Execute(executeRequestFromString("some SQL"), s.Addr(), longWait)
res, err := c.Execute(executeRequestFromString("some SQL"), s.Addr(), NO_USERNAME, NO_PASSWORD, longWait)
if err != nil {
t.Fatalf("failed to execute query: %s", err.Error())
}
@ -74,7 +80,7 @@ func Test_ServiceExecute(t *testing.T) {
}
return []*command.ExecuteResult{result}, nil
}
res, err = c.Execute(executeRequestFromString("some SQL"), s.Addr(), longWait)
res, err = c.Execute(executeRequestFromString("some SQL"), s.Addr(), NO_USERNAME, NO_PASSWORD, longWait)
if err != nil {
t.Fatalf("failed to execute: %s", err.Error())
}
@ -86,7 +92,7 @@ func Test_ServiceExecute(t *testing.T) {
time.Sleep(longWait)
return nil, nil
}
_, err = c.Execute(executeRequestFromString("some SQL"), s.Addr(), shortWait)
_, err = c.Execute(executeRequestFromString("some SQL"), s.Addr(), "", "", shortWait)
if err == nil {
t.Fatalf("failed to receive expected error")
}
@ -108,7 +114,8 @@ func Test_ServiceQuery(t *testing.T) {
go mux.Serve()
tn := mux.Listen(1) // Could be any byte value.
db := mustNewMockDatabase()
s := New(tn, db)
cred := mustNewMockCredentialStore()
s := New(tn, db, cred)
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -126,7 +133,7 @@ func Test_ServiceQuery(t *testing.T) {
}
return nil, errors.New("query failed")
}
_, err := c.Query(queryRequestFromString("SELECT * FROM foo"), s.Addr(), longWait)
_, err := c.Query(queryRequestFromString("SELECT * FROM foo"), s.Addr(), NO_USERNAME, NO_PASSWORD, longWait)
if err == nil {
t.Fatalf("client failed to report error")
}
@ -144,7 +151,7 @@ func Test_ServiceQuery(t *testing.T) {
}
return []*command.QueryRows{rows}, nil
}
res, err := c.Query(queryRequestFromString("SELECT * FROM foo"), s.Addr(), longWait)
res, err := c.Query(queryRequestFromString("SELECT * FROM foo"), s.Addr(), NO_USERNAME, NO_PASSWORD, longWait)
if err != nil {
t.Fatalf("failed to query: %s", err.Error())
}
@ -161,7 +168,7 @@ func Test_ServiceQuery(t *testing.T) {
}
return []*command.QueryRows{rows}, nil
}
res, err = c.Query(queryRequestFromString("SELECT * FROM foo"), s.Addr(), longWait)
res, err = c.Query(queryRequestFromString("SELECT * FROM foo"), s.Addr(), NO_USERNAME, NO_PASSWORD, longWait)
if err != nil {
t.Fatalf("failed to query: %s", err.Error())
}
@ -173,7 +180,7 @@ func Test_ServiceQuery(t *testing.T) {
time.Sleep(longWait)
return nil, nil
}
_, err = c.Query(queryRequestFromString("some SQL"), s.Addr(), shortWait)
_, err = c.Query(queryRequestFromString("some SQL"), s.Addr(), NO_USERNAME, NO_PASSWORD, shortWait)
if err == nil {
t.Fatalf("failed to receive expected error")
}
@ -197,7 +204,8 @@ func Test_ServiceQueryLarge(t *testing.T) {
go mux.Serve()
tn := mux.Listen(1) // Could be any byte value.
db := mustNewMockDatabase()
s := New(tn, db)
cred := mustNewMockCredentialStore()
s := New(tn, db, cred)
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -234,7 +242,7 @@ func Test_ServiceQueryLarge(t *testing.T) {
}
return []*command.QueryRows{rows}, nil
}
res, err := c.Query(queryRequestFromString("SELECT * FROM foo"), s.Addr(), longWait)
res, err := c.Query(queryRequestFromString("SELECT * FROM foo"), s.Addr(), NO_USERNAME, NO_PASSWORD, longWait)
if err != nil {
t.Fatalf("failed to query: %s", err.Error())
}

4
cluster/service_mux_test.go
Unescape Escape View File

@ -16,7 +16,7 @@ func Test_NewServiceSetGetNodeAPIAddrMuxed(t *testing.T) {
go mux.Serve()
tn := mux.Listen(1) // Could be any byte value.
s := New(tn, mustNewMockDatabase())
s := New(tn, mustNewMockDatabase(), mustNewMockCredentialStore())
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -50,7 +50,7 @@ func Test_NewServiceSetGetNodeAPIAddrMuxedTLS(t *testing.T) {
go mux.Serve()
tn := mux.Listen(1) // Could be any byte value.
s := New(tn, mustNewMockDatabase())
s := New(tn, mustNewMockDatabase(), mustNewMockCredentialStore())
if s == nil {
t.Fatalf("failed to create cluster service")
}

128
cluster/service_test.go
Unescape Escape View File

@ -2,6 +2,7 @@ package cluster
import (
"crypto/tls"
"fmt"
"net"
"os"
"testing"
@ -13,7 +14,7 @@ import (
func Test_NewServiceOpenClose(t *testing.T) {
ml := mustNewMockTransport()
s := New(ml, mustNewMockDatabase())
s := New(ml, mustNewMockDatabase(), mustNewMockCredentialStore())
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -31,7 +32,7 @@ func Test_NewServiceOpenClose(t *testing.T) {
func Test_NewServiceSetGetAPIAddr(t *testing.T) {
ml := mustNewMockTransport()
s := New(ml, mustNewMockDatabase())
s := New(ml, mustNewMockDatabase(), mustNewMockCredentialStore())
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -52,7 +53,7 @@ func Test_NewServiceSetGetAPIAddr(t *testing.T) {
func Test_NewServiceSetGetNodeAPIAddr(t *testing.T) {
ml := mustNewMockTransport()
s := New(ml, mustNewMockDatabase())
s := New(ml, mustNewMockDatabase(), mustNewMockCredentialStore())
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -97,7 +98,7 @@ func Test_NewServiceSetGetNodeAPIAddr(t *testing.T) {
func Test_NewServiceSetGetNodeAPIAddrLocal(t *testing.T) {
ml := mustNewMockTransport()
s := New(ml, mustNewMockDatabase())
s := New(ml, mustNewMockDatabase(), mustNewMockCredentialStore())
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -134,7 +135,7 @@ func Test_NewServiceSetGetNodeAPIAddrLocal(t *testing.T) {
func Test_NewServiceSetGetNodeAPIAddrTLS(t *testing.T) {
ml := mustNewMockTLSTransport()
s := New(ml, mustNewMockDatabase())
s := New(ml, mustNewMockDatabase(), mustNewMockCredentialStore())
if s == nil {
t.Fatalf("failed to create cluster service")
}
@ -170,6 +171,82 @@ func Test_NewServiceSetGetNodeAPIAddrTLS(t *testing.T) {
}
}
func Test_NewServiceTestExecuteQueryAuthNoCredentials(t *testing.T) {
ml := mustNewMockTransport()
db := mustNewMockDatabase()
// Test that for a cluster with no credential store configed
// all users are authed for both operations
var c CredentialStore = nil
c = nil
s := New(ml, db, c)
if s == nil {
t.Fatalf("failed to create cluster service")
}
if err := s.Open(); err != nil {
t.Fatalf("failed to open cluster service")
}
cl := NewClient(ml, 30*time.Second)
if err := cl.SetLocal(s.Addr(), s); err != nil {
t.Fatalf("failed to set cluster client local parameters: %s", err)
}
er := &command.ExecuteRequest{}
_, err := cl.Execute(er, s.Addr(), NO_USERNAME, NO_PASSWORD, 5*time.Second)
if err != nil {
t.Fatal(err)
}
qr := &command.QueryRequest{}
_, err = cl.Query(qr, s.Addr(), NO_USERNAME, NO_PASSWORD, 5*time.Second)
if err != nil {
t.Fatal(err)
}
}
func Test_NewServiceTestExecuteQueryAuth(t *testing.T) {
ml := mustNewMockTransport()
db := mustNewMockDatabase()
// Test that for a cluster with a credential store configed
// users with execute permissions can execute and users with
// query permissions can query and can't if they don't have those
// permissions
c := mustNewMockCredentialStoreBob()
s := New(ml, db, c)
if s == nil {
t.Fatalf("failed to create cluster service")
}
if err := s.Open(); err != nil {
t.Fatalf("failed to open cluster service")
}
cl := NewClient(ml, 30*time.Second)
if err := cl.SetLocal(s.Addr(), s); err != nil {
t.Fatalf("failed to set cluster client local parameters: %s", err)
}
er := &command.ExecuteRequest{}
_, err := cl.Execute(er, s.Addr(), "alice", "secret1", 5*time.Second)
if err != nil {
t.Fatal("alice improperly unauthorized to execute")
}
_, err = cl.Execute(er, s.Addr(), "bob", "secret1", 5*time.Second)
if err == nil {
t.Fatal("bob improperly authorized to execute")
}
qr := &command.QueryRequest{}
_, err = cl.Query(qr, s.Addr(), "bob", "secret1", 5*time.Second)
if err != nil {
fmt.Println(err)
t.Fatal("bob improperly unauthorized to query")
}
_, err = cl.Query(qr, s.Addr(), "alice", "secret1", 5*time.Second)
if err == nil {
t.Fatal("alice improperly authorized to query")
}
}
type mockTransport struct {
tn net.Listener
remoteEncrypted bool
@ -236,7 +313,13 @@ func (m *mockDatabase) Query(qr *command.QueryRequest) ([]*command.QueryRows, er
}
func mustNewMockDatabase() *mockDatabase {
return &mockDatabase{}
e := func(er *command.ExecuteRequest) ([]*command.ExecuteResult, error) {
return []*command.ExecuteResult{}, nil
}
q := func(er *command.QueryRequest) ([]*command.QueryRows, error) {
return []*command.QueryRows{}, nil
}
return &mockDatabase{executeFn: e, queryFn: q}
}
func mustCreateTLSConfig() *tls.Config {
@ -258,3 +341,36 @@ func mustCreateTLSConfig() *tls.Config {
return config
}
type mockCredentialStore struct {
HasPermOK bool
aaFunc func(username, password, perm string) bool
}
func (m *mockCredentialStore) AA(username, password, perm string) bool {
if m == nil {
return true
}
if m.aaFunc != nil {
return m.aaFunc(username, password, perm)
}
return m.HasPermOK
}
func mustNewMockCredentialStore() *mockCredentialStore {
return &mockCredentialStore{HasPermOK: true}
}
func mustNewMockCredentialStoreBob() *mockCredentialStore {
f := func(username string, password string, perm string) bool {
fmt.Println(username, password, perm)
if username == "alice" && password == "secret1" && perm == "execute" {
return true
} else if username == "bob" && password == "secret1" && perm == "query" {
return true
}
return false
}
return &mockCredentialStore{aaFunc: f}
}

6
cmd/rqlited/main.go
Unescape Escape View File

@ -102,7 +102,7 @@ func main() {
}
// Create cluster service now, so nodes will be able to learn information about each other.
clstr, err := clusterService(cfg, mux.Listen(cluster.MuxClusterHeader), str)
clstr, err := clusterService(cfg, mux.Listen(cluster.MuxClusterHeader), str, credStr)
if err != nil {
log.Fatalf("failed to create cluster service: %s", err.Error())
}
@ -310,8 +310,8 @@ func credentialStore(cfg *Config) (*auth.CredentialsStore, error) {
return cs, nil
}
func clusterService(cfg *Config, tn cluster.Transport, db cluster.Database) (*cluster.Service, error) {
c := cluster.New(tn, db)
func clusterService(cfg *Config, tn cluster.Transport, db cluster.Database, credStr *auth.CredentialsStore) (*cluster.Service, error) {
c := cluster.New(tn, db, credStr)
c.SetAPIAddr(cfg.HTTPAdv)
c.EnableHTTPS(cfg.X509Cert != "" && cfg.X509Key != "") // Conditions met for an HTTPS API

38
http/service.go
Unescape Escape View File

@ -38,6 +38,16 @@ var (
ErrUserInfoExists = errors.New("userinfo exists")
)
const (
NO_USERNAME = ""
NO_PASSSWORD = ""
)
type ResultsError interface {
Error() string
IsAuthorized() bool
}
// Database is the interface any queryable system must implement
type Database interface {
// Execute executes a slice of queries, each of which is not expected
@ -88,10 +98,10 @@ type Cluster interface {
GetNodeAPIAddr(nodeAddr string, timeout time.Duration) (string, error)
// Execute performs an Execute Request on a remote node.
Execute(er *command.ExecuteRequest, nodeAddr string, timeout time.Duration) ([]*command.ExecuteResult, error)
Execute(er *command.ExecuteRequest, nodeAddr string, username string, password string, timeout time.Duration) ([]*command.ExecuteResult, error)
// Query performs an Query Request on a remote node.
Query(qr *command.QueryRequest, nodeAddr string, timeout time.Duration) ([]*command.QueryRows, error)
Query(qr *command.QueryRequest, nodeAddr string, username string, password string, timeout time.Duration) ([]*command.QueryRows, error)
// Stats returns stats on the Cluster.
Stats() (map[string]interface{}, error)
@ -1091,7 +1101,17 @@ func (s *Service) execute(w http.ResponseWriter, r *http.Request) {
http.Error(w, ErrLeaderNotFound.Error(), http.StatusServiceUnavailable)
return
}
results, resultsErr = s.cluster.Execute(er, addr, timeout)
username, password, ok := r.BasicAuth()
if !ok {
username = ""
}
results, resultsErr = s.cluster.Execute(er, addr, username, password, timeout)
if resultsErr != nil && resultsErr.Error() == "Unauthorized" {
w.WriteHeader(http.StatusUnauthorized)
return
}
stats.Add(numRemoteExecutions, 1)
w.Header().Add(ServedByHTTPHeader, addr)
}
@ -1180,7 +1200,15 @@ func (s *Service) handleQuery(w http.ResponseWriter, r *http.Request) {
http.Error(w, ErrLeaderNotFound.Error(), http.StatusServiceUnavailable)
return
}
results, resultsErr = s.cluster.Query(qr, addr, timeout)
username, password, ok := r.BasicAuth()
if !ok {
username = ""
}
results, resultsErr = s.cluster.Query(qr, addr, username, password, timeout)
if resultsErr != nil && resultsErr.Error() == "Unauthorized" {
w.WriteHeader(http.StatusUnauthorized)
return
}
stats.Add(numRemoteQueries, 1)
w.Header().Add(ServedByHTTPHeader, addr)
}
@ -1317,7 +1345,7 @@ func (s *Service) runQueue() {
time.Sleep(retryDelay)
continue
}
_, err = s.cluster.Execute(er, addr, defaultTimeout)
_, err = s.cluster.Execute(er, addr, NO_USERNAME, NO_PASSSWORD, defaultTimeout)
if err != nil {
s.logger.Printf("execute queue write failed for sequence number %d: %s",
req.SequenceNumber, err.Error())

11
http/service_test.go
Unescape Escape View File

@ -223,9 +223,10 @@ func Test_RemoveBasicAuth(t *testing.T) {
}
func Test_NewService(t *testing.T) {
m := &MockStore{}
c := &mockClusterService{}
s := New("127.0.0.1:0", m, c, nil)
store := &MockStore{}
cluster := &mockClusterService{}
cred := &mockCredentialStore{HasPermOK: true}
s := New("127.0.0.1:0", store, cluster, cred)
if s == nil {
t.Fatalf("failed to create new service")
}
@ -1197,14 +1198,14 @@ func (m *mockClusterService) GetNodeAPIAddr(a string, t time.Duration) (string,
return m.apiAddr, nil
}
func (m *mockClusterService) Execute(er *command.ExecuteRequest, addr string, t time.Duration) ([]*command.ExecuteResult, error) {
func (m *mockClusterService) Execute(er *command.ExecuteRequest, addr string, username string, password string, t time.Duration) ([]*command.ExecuteResult, error) {
if m.executeFn != nil {
return m.executeFn(er, addr, t)
}
return nil, nil
}
func (m *mockClusterService) Query(qr *command.QueryRequest, addr string, t time.Duration) ([]*command.QueryRows, error) {
func (m *mockClusterService) Query(qr *command.QueryRequest, addr string, username string, password string, t time.Duration) ([]*command.QueryRows, error) {
if m.queryFn != nil {
return m.queryFn(qr, addr, t)
}

22
system_test/helpers.go
Unescape Escape View File

@ -594,7 +594,7 @@ func mustNodeEncryptedOnDisk(dir string, enableSingle, httpEncrypt bool, mux *tc
node.RaftAddr = node.Store.Addr()
node.ID = node.Store.ID()
clstr := cluster.New(mux.Listen(cluster.MuxClusterHeader), node.Store)
clstr := cluster.New(mux.Listen(cluster.MuxClusterHeader), node.Store, mustNewMockCredentialStore())
if err := clstr.Open(); err != nil {
panic("failed to open Cluster service)")
}
@ -883,3 +883,23 @@ func copyDir(src string, dst string) (err error) {
return
}
type mockCredentialStore struct {
HasPermOK bool
aaFunc func(username, password, perm string) bool
}
func (m *mockCredentialStore) AA(username, password, perm string) bool {
if m == nil {
return true
}
if m.aaFunc != nil {
return m.aaFunc(username, password, perm)
}
return m.HasPermOK
}
func mustNewMockCredentialStore() *mockCredentialStore {
return &mockCredentialStore{HasPermOK: true}
}

10
system_test/request_forwarding_test.go
Unescape Escape View File

@ -30,7 +30,7 @@ func Test_StoreClientSideBySide(t *testing.T) {
if exp, got := "[{}]", asJSON(res); exp != got {
t.Fatalf("unexpected results, expt %s, got %s", exp, got)
}
res, err = client.Execute(executeRequestFromString("CREATE TABLE bar (id INTEGER NOT NULL PRIMARY KEY, name TEXT)"), leaderAddr, shortWait)
res, err = client.Execute(executeRequestFromString("CREATE TABLE bar (id INTEGER NOT NULL PRIMARY KEY, name TEXT)"), leaderAddr, "", "", shortWait)
if err != nil {
t.Fatalf("failed to execute via remote: %s", err.Error())
}
@ -45,7 +45,7 @@ func Test_StoreClientSideBySide(t *testing.T) {
if exp, got := `[{"last_insert_id":1,"rows_affected":1}]`, asJSON(res); exp != got {
t.Fatalf("unexpected results, expt %s, got %s", exp, got)
}
res, err = client.Execute(executeRequestFromString(`INSERT INTO bar(name) VALUES("fiona")`), leaderAddr, shortWait)
res, err = client.Execute(executeRequestFromString(`INSERT INTO bar(name) VALUES("fiona")`), leaderAddr, "", "", shortWait)
if err != nil {
t.Fatalf("failed to execute via remote: %s", err.Error())
}
@ -68,14 +68,14 @@ func Test_StoreClientSideBySide(t *testing.T) {
t.Fatalf("unexpected results, expt %s, got %s", exp, got)
}
rows, err = client.Query(queryRequestFromString(`SELECT * FROM foo`), leaderAddr, shortWait)
rows, err = client.Query(queryRequestFromString(`SELECT * FROM foo`), leaderAddr, "", "", shortWait)
if err != nil {
t.Fatalf("failed to query via remote: %s", err.Error())
}
if exp, got := `[{"columns":["id","name"],"types":["integer","text"],"values":[[1,"fiona"]]}]`, asJSON(rows); exp != got {
t.Fatalf("unexpected results, expt %s, got %s", exp, got)
}
rows, err = client.Query(queryRequestFromString(`SELECT * FROM bar`), leaderAddr, shortWait)
rows, err = client.Query(queryRequestFromString(`SELECT * FROM bar`), leaderAddr, "", "", shortWait)
if err != nil {
t.Fatalf("failed to query via remote: %s", err.Error())
}
@ -90,7 +90,7 @@ func Test_StoreClientSideBySide(t *testing.T) {
if exp, got := `[{"error":"no such table: qux"}]`, asJSON(rows); exp != got {
t.Fatalf("unexpected results, expt %s, got %s", exp, got)
}
rows, err = client.Query(queryRequestFromString(`SELECT * FROM qux`), leaderAddr, shortWait)
rows, err = client.Query(queryRequestFromString(`SELECT * FROM qux`), leaderAddr, "", "", shortWait)
if err != nil {
t.Fatalf("failed to query via remote: %s", err.Error())
}

Write Preview
Loading…
Cancel
Save