diff --git a/api.go b/api.go index 70a27da94..47558678d 100644 --- a/api.go +++ b/api.go @@ -2753,8 +2753,8 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64 for _, flv := range flvs { fld := idx.field(flv.Field) - view, ok := fld.viewMap[flv.View] - if !ok { + view := fld.view(flv.View) + if view == nil { view, err = fld.createViewIfNotExists(flv.View) if err != nil { return err diff --git a/auth/auth.go b/auth/auth.go deleted file mode 100644 index 1eb26a73b..000000000 --- a/auth/auth.go +++ /dev/null @@ -1,25 +0,0 @@ -// Copyright 2021 Molecula Corp. All rights reserved. -package auth - -type Auth struct { - // Enable AuthZ/AuthN for featurebase server - Enable bool `toml:"enable"` - - // Application/Client ID - ClientId string `toml:"client-id"` - - // Client Secret - ClientSecret string `toml:"client-secret"` - - // Authorize URL - AuthorizeURL string `toml:"authorize-url"` - - // Token URL - TokenURL string `toml:"token-url"` - - // Group Endpoint URL - GroupEndpointURL string `toml:"group-endpoint-url"` - - // Scope URL - ScopeURL string `toml:"scope-url"` -} diff --git a/authn/authenticate.go b/authn/authenticate.go new file mode 100644 index 000000000..f4d1a8cb8 --- /dev/null +++ b/authn/authenticate.go @@ -0,0 +1,296 @@ +// Copyright 2021 Molecula Corp. All rights reserved. +package authn + +import ( + "context" + "encoding/hex" + "encoding/json" + "fmt" + "io/ioutil" + "net/http" + "time" + + "github.com/golang-jwt/jwt" + "github.com/gorilla/securecookie" + "github.com/molecula/featurebase/v2/logger" + "github.com/pkg/errors" + "golang.org/x/oauth2" +) + +type Auth struct { + logger logger.Logger + cookieName string + refreshWithin time.Duration + hashKey []byte + blockKey []byte + secure *securecookie.SecureCookie + groupEndpoint string + logoutEndpoint string + fbURL string + oAuthConfig *oauth2.Config +} + +func NewAuth(logger logger.Logger, url string, scopes []string, authUrl, tokenUrl, groupEndpoint, logout, clientID, clientSecret, hashKey, blockKey string) (*Auth, error) { + auth := &Auth{ + logger: logger, + cookieName: "molecula-chip", + refreshWithin: time.Minute * time.Duration(15), + groupEndpoint: groupEndpoint, + logoutEndpoint: logout, + fbURL: url, + oAuthConfig: &oauth2.Config{ + RedirectURL: fmt.Sprintf("%s/redirect", url), + ClientID: clientID, + ClientSecret: clientSecret, + Scopes: scopes, + Endpoint: oauth2.Endpoint{ + AuthURL: authUrl, + TokenURL: tokenUrl, + }, + }, + } + var err error + if auth.hashKey, err = decodeHex(hashKey); err != nil { + return nil, errors.Wrap(err, "decoding hash key") + } + + if auth.blockKey, err = decodeHex(blockKey); err != nil { + return nil, errors.Wrap(err, "decoding block key") + } + + auth.secure = securecookie.New(auth.hashKey, auth.blockKey) + + return auth, nil +} + +type CookieValue struct { + UserID string + UserName string + GroupMembership []Group + Token *oauth2.Token +} + +type Groups struct { + Groups []Group `json:"value"` +} + +type Group struct { + UserID string + GroupID string `json:"id"` + GroupName string `json:"displayName"` +} + +type UserInfo struct { + UserID string `json:"userid"` + UserName string `json:"username"` +} + +func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) ([]Group, error) { + cookie, err := a.readCookie(w, r) + if err != nil { + http.Redirect(w, r, "/signin", http.StatusTemporaryRedirect) + return nil, err + } + if cookie.Token.Expiry.Before(time.Now().Add(a.refreshWithin)) { + err = a.refreshToken(w, cookie) + if err != nil { + a.logger.Errorf("refreshing access token: ", err) + if cookie.Token.Expiry.Before(time.Now()) { + http.Redirect(w, r, "/signin", http.StatusTemporaryRedirect) + return nil, err + } + } + } + if len(cookie.GroupMembership) == 0 { + return nil, errors.New("user is not part of any groups in identity provider") + } + return cookie.GroupMembership, nil + +} + +func (a *Auth) Login(w http.ResponseWriter, r *http.Request) { + authUrl := a.oAuthConfig.AuthCodeURL(a.oAuthConfig.Endpoint.AuthURL) + http.Redirect(w, r, authUrl, http.StatusTemporaryRedirect) +} + +func (a *Auth) Logout(w http.ResponseWriter, r *http.Request) { + newCookie := a.getEmptyCookie() + http.SetCookie(w, newCookie) + redirect := fmt.Sprintf("%s?post_logout_redirect_uri=%s/", a.logoutEndpoint, a.fbURL) + http.Redirect(w, r, redirect, http.StatusTemporaryRedirect) +} + +// Gets user information from dP and sets a secure cookie +func (a *Auth) Redirect(w http.ResponseWriter, r *http.Request) { + code := r.FormValue("code") + token, err := a.getToken(code) + if err != nil { + http.Error(w, "Bad Request: 400", http.StatusBadRequest) + return + } + + cv, err := a.newCookieValue(token) + if err != nil || cv == nil { + http.Error(w, "Bad Request: 400", http.StatusBadRequest) + return + } + + a.setCookie(w, cv) + http.Redirect(w, r, "/", http.StatusTemporaryRedirect) +} + +func (a *Auth) GetUserInfo(w http.ResponseWriter, r *http.Request) *UserInfo { + var resp UserInfo + cookie, err := a.readCookie(w, r) + if err != nil { + //add logging + return &resp + } + resp.UserID = cookie.UserID + resp.UserName = cookie.UserName + return &resp + +} + +func (a *Auth) getToken(code string) (*oauth2.Token, error) { + token, err := a.oAuthConfig.Exchange(context.Background(), code) + if err != nil { + return nil, errors.Wrap(err, "exchanging auth code for token") + } + return token, nil +} + +func (a *Auth) newCookieValue(token *oauth2.Token) (*CookieValue, error) { + if token == nil { + return nil, errors.New("baking cookie due to nil token") + } + if token.AccessToken == "" { + return nil, errors.New("no access token provided") + } + accessParsed, err := jwt.Parse(token.AccessToken, nil) + if accessParsed == nil || accessParsed.Claims == nil { + return nil, errors.Wrap(err, "parsing jwt claims from access tokens") + } + claims := accessParsed.Claims.(jwt.MapClaims) + + groups, err := a.getGroupMembership(token) + if err != nil { + return nil, errors.Wrap(err, "getting group membership") + } + // not needed at this point in the logic and makes the encoded cookie too large + token.AccessToken = "" + return &CookieValue{ + UserID: claims["oid"].(string), + UserName: claims["name"].(string), + GroupMembership: groups.Groups, + Token: token, + }, nil +} + +func (a *Auth) getGroupMembership(token *oauth2.Token) (Groups, error) { + var groups Groups + var bearer = fmt.Sprintf("Bearer %s", token.AccessToken) + req, err := http.NewRequest("GET", a.groupEndpoint, nil) + if err != nil { + return groups, errors.Wrap(err, "creating new request to group endpoint") + } + + req.Header.Add("Authorization", bearer) + client := &http.Client{} + response, err := client.Do(req) + if err != nil { + return groups, errors.Wrap(err, "getting group membership info") + } + + defer response.Body.Close() + rawGroups, err := ioutil.ReadAll(response.Body) + if err != nil { + return groups, errors.Wrap(err, "failed reading group membership response") + } + + if err = json.Unmarshal(rawGroups, &groups); err != nil { + return groups, errors.Wrap(err, "failed unmarshalling group membership response") + } + + return groups, nil +} + +func (a *Auth) readCookie(w http.ResponseWriter, r *http.Request) (*CookieValue, error) { + cookie, err := r.Cookie(a.cookieName) + if err != nil { + return nil, errors.Wrap(err, "cookie not found") + } + + var value CookieValue + err = a.secure.Decode(a.cookieName, cookie.Value, &value) + if err != nil { + newCookie := a.getEmptyCookie() + http.SetCookie(w, newCookie) + return nil, errors.Wrap(err, "decoding cookie") + } + + return &value, nil +} + +func (a *Auth) setCookie(w http.ResponseWriter, cookie *CookieValue) error { + encoded, err := a.secure.Encode(a.cookieName, cookie) + if err != nil { + return errors.Wrap(err, "encoding CookieValue") + + } + newCookie := &http.Cookie{ + Name: a.cookieName, + Value: encoded, + Path: "/", + Secure: true, + HttpOnly: true, + SameSite: http.SameSiteStrictMode, + Expires: cookie.Token.Expiry, + } + http.SetCookie(w, newCookie) + return nil +} + +func (a *Auth) refreshToken(w http.ResponseWriter, cookie *CookieValue) error { + if cookie.Token.RefreshToken == "" { + return errors.New("no refresh token found, check auth scopes to see if refresh tokens are being provided by your IdP.") + } + tokenSource := a.oAuthConfig.TokenSource(context.Background(), cookie.Token) + newToken, err := tokenSource.Token() + if err != nil { + return errors.Wrap(err, "refreshing token") + } + + if newToken.Expiry != cookie.Token.Expiry { + cv, err := a.newCookieValue(newToken) + if err != nil { + errors.Wrap(err, "setting cookie") + } + + a.setCookie(w, cv) + } + + return nil +} + +func decodeHex(hexstr string) ([]byte, error) { + data, err := hex.DecodeString(hexstr) + if err != nil { + return nil, errors.Wrap(err, "decoding hex string to byte slice") + } + if len(data) != 32 { + return nil, errors.Wrap(err, "invalid key length") + } + return data, nil +} + +func (a *Auth) getEmptyCookie() *http.Cookie { + return &http.Cookie{ + Name: a.cookieName, + Value: "", + Path: "/", + Secure: true, + HttpOnly: true, + SameSite: http.SameSiteStrictMode, + } +} diff --git a/authn/authenticate_internal_test.go b/authn/authenticate_internal_test.go new file mode 100644 index 000000000..8df8c3b03 --- /dev/null +++ b/authn/authenticate_internal_test.go @@ -0,0 +1,118 @@ +package authn + +import ( + "net/http/httptest" + "os" + "strings" + "testing" + "time" + + "github.com/molecula/featurebase/v2/logger" + "golang.org/x/oauth2" +) + +func TestAuth(t *testing.T) { + var ( + ClientId = "e9088663-eb08-41d7-8f65-efb5f54bbb71" + ClientSecret = "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF" + AuthorizeURL = "https://login.microsoftonline.com/4a137d66-d161-4ae4-b1e6-07e9920874b8/oauth2/v2.0/authorize" + TokenURL = "https://login.microsoftonline.com/4a137d66-d161-4ae4-b1e6-07e9920874b8/oauth2/v2.0/token" + GroupEndpointURL = "https://graph.microsoft.com/v1.0/me/transitiveMemberOf/microsoft.graph.group?$count=true" + LogoutURL = "https://login.microsoftonline.com/common/oauth2/v2.0/logout" + Scopes = []string{"https://graph.microsoft.com/.default", "offline_access"} + Key = "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF" + ShortKey = "DEADBEEFD" + ) + + a, err := NewAuth( + logger.NewStandardLogger(os.Stdout), + "http://localhost:10101/", + Scopes, + AuthorizeURL, + TokenURL, + GroupEndpointURL, + LogoutURL, + ClientId, + ClientSecret, + Key, + Key, + ) + if err != nil { + t.Errorf("building auth object%s", err) + } + tokenNoAT := oauth2.Token{ + TokenType: "Bearer", + RefreshToken: "abcdef", + Expiry: time.Now().Add(time.Hour), + } + tokenAT := oauth2.Token{ + TokenType: "Bearer", + RefreshToken: "abcdef", + AccessToken: "aasdf", + Expiry: time.Now().Add(time.Hour), + } + grp := Group{ + UserID: "snowstorm", + GroupID: "abcd123-A", + GroupName: "Romantic Painters", + } + validCV := CookieValue{ + UserID: "snowstorm", + UserName: "J.M.W. Turner", + GroupMembership: []Group{grp}, + Token: &tokenAT, + } + + t.Run("SetCookie", func(t *testing.T) { + w := httptest.NewRecorder() + err := a.setCookie(w, &validCV) + if err != nil { + t.Errorf("expected no errors, got: %v", err) + } + + if w.Result().Cookies()[0].Value == "" { + t.Errorf("expected some value, got: %+v", w.Result().Cookies()[0].Value) + } + if w.Result().Cookies()[0].Path != "/" { + t.Errorf("expected path to be /, got: %+v", w.Result().Cookies()[0].Path) + } + }) + t.Run("GetEmptyCookie", func(t *testing.T) { + c := a.getEmptyCookie() + if c.Value != "" { + t.Errorf("expected empty cookie, got: %+v", c.Value) + } + }) + t.Run("KeyLength", func(t *testing.T) { + _, err := NewAuth( + logger.NewStandardLogger(os.Stdout), + "http://localhost:10101/", + Scopes, + AuthorizeURL, + TokenURL, + GroupEndpointURL, + LogoutURL, + ClientId, + ClientSecret, + Key, + ShortKey, + ) + if err == nil || !strings.Contains(err.Error(), "decoding block key") { + t.Errorf("expected error decoding block key got: %v", err) + } + }) + t.Run("NewCookieValue-BadAccessToken", func(t *testing.T) { + _, err := a.newCookieValue(&tokenAT) + if err == nil || !strings.Contains(err.Error(), "jwt claims") { + t.Errorf("expected failure regarding jwt claims, got: %v", err) + } + + }) + t.Run("CookieValue-NoAccessToken", func(t *testing.T) { + _, err := a.newCookieValue(&tokenNoAT) + if err == nil || !strings.Contains(err.Error(), "access token") { + t.Errorf("expected failure regarding access token, got: %v", err) + } + }) + +} diff --git a/authz/authorization.go b/authz/authorization.go new file mode 100644 index 000000000..11bc9faac --- /dev/null +++ b/authz/authorization.go @@ -0,0 +1,120 @@ +// Copyright 2017 Pilosa Corp. +// +// 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 authz + +import ( + "fmt" + "io" + "io/ioutil" + + "github.com/molecula/featurebase/v2/authn" + + "gopkg.in/yaml.v2" +) + +type GroupPermissions struct { + Permissions map[string]map[string]string `yaml:"user-groups"` + Admin string `yaml:"admin"` +} + +func (p *GroupPermissions) ReadPermissionsFile(permsFile io.Reader) (err error) { + permsData, err := ioutil.ReadAll(permsFile) + + if err != nil { + return fmt.Errorf("reading permissions failed with error: %s", err) + } + + err = yaml.UnmarshalStrict(permsData, &p) + if err != nil { + return fmt.Errorf("unmarshalling permissions failed with error: %s", err) + } + + return +} + +func (p *GroupPermissions) GetPermissions(groups []authn.Group, index string) (permission string, errors error) { + + if admin := p.IsAdmin(groups); admin { + return "admin", nil + } + + allPermissions := map[string]bool{ + "write": false, + "read": false, + } + + if len(groups) == 0 { + return "", fmt.Errorf("user is not part of any groups in identity provider") + } + + var groupsDenied []string + for _, group := range groups { + if _, ok := p.Permissions[group.GroupID]; ok { + if perm, ok := p.Permissions[group.GroupID][index]; ok { + allPermissions[perm] = true + } else { + return "", fmt.Errorf("user %s does not have permission to index %s", group.UserID, index) + } + } else { + groupsDenied = append(groupsDenied, group.GroupID) + } + } + + if len(groupsDenied) == len(groups) { + return "", fmt.Errorf("group(s) %s does not have permission to FeatureBase", groupsDenied) + } + + if allPermissions["write"] { + return "write", nil + } else if allPermissions["read"] { + return "read", nil + } else { + return "", fmt.Errorf("no permissions found") + } +} + +func (p *GroupPermissions) IsAdmin(groups []authn.Group) bool { + for _, group := range groups { + if p.Admin == group.GroupID { + return true + } + } + return false +} + +func (p *GroupPermissions) GetAuthorizedIndexList(groups []authn.Group, desiredPermission string) (indexList []string) { + // if user is admin, find all indexes in permissions file and return them + if admin := p.IsAdmin(groups); admin { + for groupId := range p.Permissions { + for index := range p.Permissions[groupId] { + indexList = append(indexList, index) + } + } + return indexList + } + + for _, group := range groups { + if _, ok := p.Permissions[group.GroupID]; ok { + for index, permission := range p.Permissions[group.GroupID] { + if permission == desiredPermission { + indexList = append(indexList, index) + } else if permission == "write" && desiredPermission == "read" { + indexList = append(indexList, index) + } + } + } + } + return indexList +} diff --git a/authz/authorization_test.go b/authz/authorization_test.go new file mode 100644 index 000000000..bfda894a9 --- /dev/null +++ b/authz/authorization_test.go @@ -0,0 +1,318 @@ +// Copyright 2017 Pilosa Corp. +// +// 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 authz_test + +import ( + "fmt" + "reflect" + "sort" + "strings" + "testing" + + "github.com/molecula/featurebase/v2/authn" + "github.com/molecula/featurebase/v2/authz" +) + +func TestAuth_ReadPermissionsFile(t *testing.T) { + + singleInput := `user-groups: + "dca35310-ecda-4f23-86cd-876aee55906b": + "test": "read" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + multiInput := `user-groups: + "dca35310-ecda-4f23-86cd-876aee55906b": + "test": "read" + "test2": "write" + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "write" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + singlePermission := authz.GroupPermissions{ + Permissions: map[string]map[string]string{ + "dca35310-ecda-4f23-86cd-876aee55906b": {"test": "read"}, + }, + Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", + } + + multiPermission := authz.GroupPermissions{ + Permissions: map[string]map[string]string{ + "dca35310-ecda-4f23-86cd-876aee55906b": {"test": "read", "test2": "write"}, + "dca35310-ecda-4f23-86cd-876aee559900": {"test": "write"}}, + Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", + } + + tests := []struct { + input string + output authz.GroupPermissions + }{ + {singleInput, singlePermission}, + {multiInput, multiPermission}, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + permFile := strings.NewReader(test.input) + + var p authz.GroupPermissions + err := p.ReadPermissionsFile(permFile) + if err != nil { + t.Fatalf("readPermissionsFile error: %s", err) + } + + if !reflect.DeepEqual(p, test.output) { + t.Fatalf("expected output %s, but got %s", test.output, p) + } + }, + ) + } +} + +func TestAuth_GetPermissions(t *testing.T) { + + // initializes different example of permissions file in yaml + permissions1 := `"user-groups": + "dca35310-ecda-4f23-86cd-876aee55906b": + "test": "read" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions2 := `"user-groups": + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "write" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions3 := `"user-groups": + "dca35310-ecda-4f23-86cd-876aee55906b": + "test": "write" + "test2": "read" + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "read" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions4 := `"user-groups": + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + // initializes groups that are returned from identity provider + groupName := "name" + userId := "user-id" + groupsList1 := []authn.Group{} + groupsList2 := []authn.Group{{ + UserID: userId, + GroupID: "fake-group", + GroupName: groupName}} + groupsList3 := []authn.Group{ + {UserID: userId, GroupID: "dca35310-ecda-4f23-86cd-876aee55906b", GroupName: groupName}, + {UserID: userId, GroupID: "dca35310-ecda-4f23-86cd-876aee559900", GroupName: groupName}, + } + groupsList4 := []authn.Group{{UserID: userId, GroupID: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", GroupName: groupName}} + + tests := []struct { + yamlData string + groups []authn.Group + index string + userAccess string + err string + }{ + { + permissions1, + groupsList1, + "test", + "", + "user is not part of any groups in identity provider", + }, + { + permissions1, + groupsList3, + "test1", + "", + "does not have permission to index", + }, + { + permissions2, + groupsList2, + "test", + "", + "does not have permission to FeatureBase", + }, + { + permissions1, + groupsList3, + "test", + "read", + "", + }, + { + permissions2, + groupsList3, + "test", + "write", + "", + }, + { + permissions3, + groupsList4, + "test", + "admin", + "", + }, + { + permissions4, + groupsList3, + "test", + "", + "no permissions found", + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + + permFile := strings.NewReader(test.yamlData) + + var p authz.GroupPermissions + if err := p.ReadPermissionsFile(permFile); err != nil { + t.Errorf("Error: %s", err) + } + + p1, err := p.GetPermissions(test.groups, test.index) + + if p1 != test.userAccess { + t.Errorf("expected permission to be %s, but got %s", test.userAccess, p1) + } + + if err != nil { + if !strings.Contains(err.Error(), test.err) { + t.Errorf("expected error to contain %s, but got %s", test.err, err.Error()) + } + } + + }) + } +} + +func TestAuth_IsAdmin(t *testing.T) { + + group1 := []authn.Group{ + {UserID: "admin-user-id", GroupID: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", GroupName: "admin-group"}, + } + + group2 := []authn.Group{ + {UserID: "user-id", GroupID: "dca35310-ecda-4f23-86cd-876aee55906b", GroupName: "group-name"}, + } + + groupPermissions := authz.GroupPermissions{ + Permissions: map[string]map[string]string{ + "dca35310-ecda-4f23-86cd-876aee55906b": {"test": "write"}, + }, + Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", + } + + tests := []struct { + groups []authn.Group + groupPermissions authz.GroupPermissions + output bool + }{ + { + group1, groupPermissions, true, + }, + { + group2, groupPermissions, false, + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + p := test.groupPermissions + resp := p.IsAdmin(test.groups) + if resp != test.output { + t.Errorf("expected %t, but got %t", test.output, resp) + } + }) + } +} + +func TestAuth_GetAuthorizedIndexList(t *testing.T) { + + group1 := []authn.Group{ + {UserID: "user-id", GroupID: "dca35310-ecda-4f23-86cd-876aee55906b", GroupName: "group-name"}, + } + + group2 := []authn.Group{ + {UserID: "admin-user-id", GroupID: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", GroupName: "admin-group"}, + } + + group3 := []authn.Group{ + {UserID: "user-id", GroupID: "dca35310-ecda-4f23-86cd-876aee559900", GroupName: "group-name"}, + } + + p := authz.GroupPermissions{ + Permissions: map[string]map[string]string{ + "dca35310-ecda-4f23-86cd-876aee55906b": { + "test1": "read", + "test2": "write", + }, + "dca35310-ecda-4f23-86cd-876aee559900": { + "test3": "read", + }, + }, + Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", + } + + tests := []struct { + groups []authn.Group + permission string + output []string + }{ + { + group1, + "read", + []string{"test1", "test2"}, + }, + { + group1, + "write", + []string{"test2"}, + }, + { + group3, + "write", + nil, + }, + { + group2, + "read", + []string{"test1", "test2", "test3"}, + }, + { + group2, + "write", + []string{"test1", "test2", "test3"}, + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + + indexList := p.GetAuthorizedIndexList(test.groups, test.permission) + sort.Strings(indexList) + + if !reflect.DeepEqual(indexList, test.output) { + t.Errorf("expected %s, but got %s", test.output, indexList) + } + }) + } + +} diff --git a/cmd/keygen.go b/cmd/keygen.go new file mode 100644 index 000000000..9a4faa940 --- /dev/null +++ b/cmd/keygen.go @@ -0,0 +1,28 @@ +// Copyright 2021 Molecula Corp. All rights reserved. +package cmd + +import ( + "context" + "io" + + "github.com/molecula/featurebase/v2/ctl" + "github.com/spf13/cobra" +) + +func newKeygenCommand(stdin io.Reader, stdout io.Writer, stderr io.Writer) *cobra.Command { + cmd := ctl.NewKeygenCommand(stdin, stdout, stderr) + ccmd := &cobra.Command{ + Use: "keygen", + Short: "Generate keys for authentication.", + Long: ` +Generate hash and block keys to configure FeatureBase for Authentication. +`, + RunE: func(c *cobra.Command, args []string) error { + return cmd.Run(context.Background()) + }, + } + + flags := ccmd.Flags() + flags.IntVarP(&cmd.KeyLength, "length", "l", 32, "length of keys to produce") + return ccmd +} diff --git a/cmd/root.go b/cmd/root.go index 4ea6a30e0..c164bed97 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -6,7 +6,7 @@ import ( "io" "strings" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" "github.com/spf13/cobra" "github.com/spf13/pflag" "github.com/spf13/viper" @@ -62,6 +62,8 @@ at https://docs.molecula.cloud/. rc.AddCommand(newRBFCommand(stdin, stdout, stderr)) rc.AddCommand(newServeCmd(stdin, stdout, stderr)) rc.AddCommand(newHolderCmd(stdin, stdout, stderr)) + rc.AddCommand(newHolderCmd(stdin, stdout, stderr)) + rc.AddCommand(newKeygenCommand(stdin, stdout, stderr)) rc.SetOutput(stderr) return rc diff --git a/ctl/keygen.go b/ctl/keygen.go new file mode 100644 index 000000000..06cc797ad --- /dev/null +++ b/ctl/keygen.go @@ -0,0 +1,31 @@ +// Copyright 2021 Molecula Corp. All rights reserved. +package ctl + +import ( + "context" + "fmt" + "io" + + "github.com/gorilla/securecookie" + pilosa "github.com/molecula/featurebase/v2" +) + +// Keygen represents a command for generating crytographic keys. +type KeygenCommand struct { + CmdIO *pilosa.CmdIO + KeyLength int +} + +// NewKeygen returns a new instance of Keygen. +func NewKeygenCommand(stdin io.Reader, stdout, stderr io.Writer) *KeygenCommand { + return &KeygenCommand{ + CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), + } +} + +// Run keys to use for authentication . +func (kg *KeygenCommand) Run(_ context.Context) error { + fmt.Printf("hash-key = \"%+x\"\n", securecookie.GenerateRandomKey(kg.KeyLength)) + fmt.Printf("block-key = \"%+x\"\n", securecookie.GenerateRandomKey(kg.KeyLength)) + return nil +} diff --git a/ctl/rbf_pages.go b/ctl/rbf_pages.go index 2e831892a..ca3484f31 100644 --- a/ctl/rbf_pages.go +++ b/ctl/rbf_pages.go @@ -69,9 +69,9 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { // Print one line for each page. for pgno, info := range infos { + fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) switch info := info.(type) { case *rbf.MetaPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "meta") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", "") @@ -79,7 +79,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "pageN=%d,walid=%d,rootrec=%d,freelist=%d\n", info.PageN, info.WALID, info.RootRecordPageNo, info.FreelistPageNo) case *rbf.RootRecordPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "rootrec") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", "") @@ -87,7 +86,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "next=%d\n", info.Next) case *rbf.LeafPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "leaf") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree)) @@ -95,7 +93,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "flags=x%x,celln=%d\n", info.Flags, info.CellN) case *rbf.BranchPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "branch") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree)) @@ -103,7 +100,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "flags=x%x,celln=%d\n", info.Flags, info.CellN) case *rbf.BitmapPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "bitmap") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree)) @@ -111,7 +107,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "-\n") case *rbf.FreePageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "free") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", "") @@ -119,7 +114,7 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "-\n") default: - panic(fmt.Sprintf("unexpected page info type %T", info)) + fmt.Fprintf(cmd.Stdout, "unknown [%T]\n", info) } } diff --git a/ctl/server.go b/ctl/server.go index 5379ce0d3..3a4ccf463 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -86,7 +86,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.BoolVar(&srv.Config.Storage.FsyncEnabled, "storage.fsync", true, "enable fsync fully safe flush-to-disk") // RowcacheOn - flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "turn on the rowcache for all backends (may speed some queries)") + flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "Do not use, permanently disabled. Flag exists for backwards compatibility and will be removed.") // RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions. srv.Config.RBFConfig.DefineFlags(flags) @@ -117,6 +117,10 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVar(&srv.Config.Auth.AuthorizeURL, "auth.authorize-url", srv.Config.Auth.AuthorizeURL, "Identity Provider's Authorize URL.") flags.StringVar(&srv.Config.Auth.TokenURL, "auth.token-url", srv.Config.Auth.TokenURL, "Identity Provider's Token URL.") flags.StringVar(&srv.Config.Auth.GroupEndpointURL, "auth.group-endpoint-url", srv.Config.Auth.GroupEndpointURL, "Identity Provider's Group endpoint URL.") - flags.StringVar(&srv.Config.Auth.ScopeURL, "auth.scope-url", srv.Config.Auth.ScopeURL, "Identity Provider's Scope URL.") + flags.StringVar(&srv.Config.Auth.LogoutURL, "auth.logout-url", srv.Config.Auth.LogoutURL, "Identity Provider's Logout URL.") + flags.StringSliceVar(&srv.Config.Auth.Scopes, "auth.scopes", srv.Config.Auth.Scopes, "Comma separated list of scopes obtained from IdP") + flags.StringVar(&srv.Config.Auth.HashKey, "auth.hash-key", srv.Config.Auth.HashKey, "First Secret for Auth.") + flags.StringVar(&srv.Config.Auth.BlockKey, "auth.block-key", srv.Config.Auth.BlockKey, "Second Secret for Auth.") + flags.StringVar(&srv.Config.Auth.PermissionsFile, "auth.permissions", srv.Config.Auth.PermissionsFile, "Permissions' file with group authorization.") } diff --git a/executor.go b/executor.go index 226759b55..5e03e1a72 100644 --- a/executor.go +++ b/executor.go @@ -51,6 +51,9 @@ type executor struct { Node *topology.Node Cluster *cluster + // how many jobs the work queue has seen + workCounter uint64 + // Client used for remote requests. client InternalQueryClient @@ -61,6 +64,7 @@ type executor struct { workMu sync.RWMutex workersWG sync.WaitGroup workerPoolSize int + currentWorkers int64 work chan job // Maximum per-request memory usage (Extract() only) @@ -128,15 +132,61 @@ func newExecutor(opts ...executorOption) *executor { e.work = make(chan job, e.workerPoolSize) _ = testhook.Opened(NewAuditor(), e, nil) for i := 0; i < e.workerPoolSize; i++ { - e.workersWG.Add(1) - go func() { - defer e.workersWG.Done() - worker(e.work) - }() + e.addWorker() } + go func() { + // background task: every so often, check to see whether we have + // work in the queue but none has been taken for a while. if so, we + // need more workers. + prev := atomic.LoadUint64(&e.workCounter) + periodic := time.NewTicker(50 * time.Millisecond) + defer periodic.Stop() + running := true + idle := 0 + for running { + <-periodic.C + func() { + e.workMu.RLock() + defer e.workMu.RUnlock() + if e.shutdown { + running = false + return + } + if len(e.work) == 0 { + idle++ + if idle > 10 && atomic.LoadInt64(&e.currentWorkers) > int64(e.workerPoolSize*2) { + select { + case e.work <- job{idleHands: true}: + // we closed an excess worker + default: + // somehow between our test above and now the work + // queue FILLED UP and we stoically accept this + } + idle = 0 + } + return + } + next := atomic.LoadUint64(&e.workCounter) + if next == prev { + e.addWorker() + } + prev = next + }() + } + }() return e } +func (e *executor) addWorker() { + e.workersWG.Add(1) + atomic.AddInt64(&e.currentWorkers, 1) + go func() { + defer e.workersWG.Done() + e.worker(e.work) + atomic.AddInt64(&e.currentWorkers, -1) + }() +} + func (e *executor) Close() error { e.workMu.Lock() defer e.workMu.Unlock() @@ -4493,7 +4543,11 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, return nil, err } defer finisher(&err0) - return frag.row(tx, rowID) + row, err := frag.row(tx, rowID) + if qcx.write && err == nil { + row = row.Clone() + } + return row, err } // If no quantum exists then return an empty bitmap. @@ -4532,15 +4586,21 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, if len(rows) == 0 { return &Row{}, nil } else if len(rows) == 1 { + if qcx.write { + return rows[0].Clone(), nil + } return rows[0], nil } row := rows[0].Union(rows[1:]...) + if qcx.write { + row = row.Clone() + } return row, nil } // executeRowBSIGroupShard executes a range(bsiGroup) call for a local shard. -func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ *Row, err0 error) { +func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (cloneable *Row, err0 error) { span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeRowBSIGroupShard") defer span.Finish() @@ -4572,6 +4632,11 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index return nil, err } defer finisher(&err0) + defer func() { + if qcx.write && cloneable != nil { + cloneable = cloneable.Clone() + } + }() // EQ null _exists - frag.NotNull() // NEQ null frag.NotNull() @@ -4822,6 +4887,9 @@ func (e *executor) executeNotShard(ctx context.Context, qcx *Qcx, index string, if existenceRow, err = existenceFrag.row(tx, 0); err != nil { return nil, err } + if qcx.write { + existenceRow = existenceRow.Clone() + } } // the finishers returned by a write tx, which we might be in if there's // a higher-level write in this call OR ANY OTHER CALL, are safe to @@ -5915,10 +5983,15 @@ type job struct { ctx context.Context memoryAvailable *int64 // shared, atomic value resultChan chan mapResponse + idleHands bool } -func worker(work chan job) { +func (e *executor) worker(work chan job) { for j := range work { + atomic.AddUint64(&e.workCounter, 1) + if j.idleHands { + return + } // Skip out early if the context is done, but still send // an ack so mapperLocal can be sure we aren't about to // work on something it sent us. diff --git a/field.go b/field.go index 2aa865a1f..d325c477e 100644 --- a/field.go +++ b/field.go @@ -705,8 +705,11 @@ func (f *Field) cacheBitDepth(bd uint64) error { f.mu.Lock() defer f.mu.Unlock() - f.options.BitDepth = bd - if bsig != nil { + if f.options.BitDepth < bd { + f.options.BitDepth = bd + } + + if bsig != nil && bsig.BitDepth < bd { bsig.BitDepth = bd } diff --git a/fragment.go b/fragment.go index 83514edad..8dfa6749c 100644 --- a/fragment.go +++ b/fragment.go @@ -218,6 +218,8 @@ func newFragment(holder *Holder, spec fragSpec, shard uint64, flags byte) *fragm func (f *fragment) cachePath() string { return f.path() + cacheExt } func (f *fragment) bitDepth() (uint64, error) { + f.mu.RLock() + defer f.mu.RUnlock() tx, err := f.holder.BeginTx(false, f.idx, f.shard) if err != nil { return 0, errors.Wrapf(err, "beginning new tx(false, %s, %d)", f.index(), f.shard) @@ -593,8 +595,8 @@ func (f *fragment) mutexCheck(tx Tx, details bool, limit int) (map[uint64][]uint // row returns a row by ID. func (f *fragment) row(tx Tx, rowID uint64) (*Row, error) { - f.mu.Lock() - defer f.mu.Unlock() + f.mu.RLock() + defer f.mu.RUnlock() return f.unprotectedRow(tx, rowID) } @@ -937,9 +939,12 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e return changed, nil } -// unprotectedClearBlock clears all rows for a given block. +// clearBlock clears all rows for a given block. // This updates both the on-disk storage and the in-cache bitmap. -func (f *fragment) unprotectedClearBlock(tx Tx, block int) (changed bool, err error) { +func (f *fragment) clearBlock(tx Tx, block int) (changed bool, err error) { + f.mu.Lock() + defer f.mu.Unlock() + firstRow := uint64(block * HashBlockSize) var wp *io.Writer if f.storage != nil { @@ -2708,20 +2713,24 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep func (f *fragment) importRoaring(ctx context.Context, tx Tx, data []byte, clear bool) error { span, ctx := tracing.StartSpanFromContext(ctx, "fragment.importRoaring") defer span.Finish() - span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.AcquireFragmentLock") - f.mu.Lock() - defer f.mu.Unlock() - span.Finish() - return f.unprotectedImportRoaring(ctx, tx, data, clear) + rowSet, updateCache, err := f.doImportRoaring(ctx, tx, data, clear) + if err != nil { + return errors.Wrap(err, "doImportRoaring") + } + if updateCache { + return f.updateCachePostImport(ctx, rowSet) + } + return nil } -func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) error { +func (f *fragment) doImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) (map[uint64]int, bool, error) { + f.mu.RLock() + defer f.mu.RUnlock() rowSize := uint64(1 << shardVsContainerExponent) span, ctx := tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits") + defer span.Finish() - useRowCache := storage.RowCacheEnabled() - var changed int var rowSet map[uint64]int var wp *io.Writer if f.storage != nil { @@ -2734,37 +2743,37 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b return err } - changed, rowSet, err = tx.ImportRoaringBits(f.index(), f.field(), f.view(), f.shard, rit, clear, true, rowSize) + _, rowSet, err = tx.ImportRoaringBits(f.index(), f.field(), f.view(), f.shard, rit, clear, true, rowSize) return err }) - span.Finish() if err != nil { - return err + return nil, false, err } updateCache := f.CacheType != CacheTypeNone + return rowSet, updateCache, err +} + +func (f *fragment) updateCachePostImport(ctx context.Context, rowSet map[uint64]int) error { + f.mu.Lock() + defer f.mu.Unlock() anyChanged := false for rowID, changes := range rowSet { if changes == 0 { continue } - if useRowCache && f.rowCache != nil { - f.rowCache.Add(rowID, nil) - } - if updateCache { - anyChanged = true - if changes < 0 { - absChanges := uint64(-1 * changes) - if absChanges <= f.cache.Get(rowID) { - f.cache.BulkAdd(rowID, f.cache.Get(rowID)-absChanges) - } else { - f.cache.BulkAdd(rowID, 0) - } + anyChanged = true + if changes < 0 { + absChanges := uint64(-1 * changes) + if absChanges <= f.cache.Get(rowID) { + f.cache.BulkAdd(rowID, f.cache.Get(rowID)-absChanges) } else { - f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes)) + f.cache.BulkAdd(rowID, 0) } + } else { + f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes)) } } // we only set this if we need to update the cache @@ -2772,26 +2781,18 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b f.cache.Invalidate() } - span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.incrementOpN") - - f.incrementOpN(changed) - - span.Finish() return nil } // importRoaringOverwrite overwrites the specified block with the provided data. func (f *fragment) importRoaringOverwrite(ctx context.Context, tx Tx, data []byte, block int) error { - f.mu.Lock() - defer f.mu.Unlock() - // Clear the existing data from fragment block. - if _, err := f.unprotectedClearBlock(tx, block); err != nil { + if _, err := f.clearBlock(tx, block); err != nil { return errors.Wrapf(err, "clearing block: %d", block) } // Union the new block data with the fragment data. - return f.unprotectedImportRoaring(ctx, tx, data, false) + return f.importRoaring(ctx, tx, data, false) } // incrementOpN increase the operation count by one. diff --git a/go.mod b/go.mod index add6082e3..24e125060 100644 --- a/go.mod +++ b/go.mod @@ -19,12 +19,14 @@ require ( github.com/fsnotify/fsnotify v1.4.9 // indirect github.com/go-test/deep v1.0.7 github.com/gogo/protobuf v1.3.2 + github.com/golang-jwt/jwt v3.2.2+incompatible github.com/golang/protobuf v1.3.3 github.com/google/go-cmp v0.5.5 github.com/google/uuid v1.1.4 // indirect github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00 // indirect github.com/gorilla/handlers v1.3.0 github.com/gorilla/mux v1.7.0 + github.com/gorilla/securecookie v1.1.1 github.com/improbable-eng/grpc-web v0.13.0 github.com/lib/pq v1.8.0 github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b @@ -52,9 +54,10 @@ require ( golang.org/x/exp v0.0.0-20201008143054-e3b2a7f2fdc7 golang.org/x/mod v0.4.2 golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d // indirect + golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c google.golang.org/grpc v1.28.0 - gopkg.in/yaml.v2 v2.3.0 // indirect + gopkg.in/yaml.v2 v2.3.0 modernc.org/mathutil v1.0.0 modernc.org/strutil v1.0.0 sigs.k8s.io/yaml v1.2.0 // indirect diff --git a/go.sum b/go.sum index 4224f965d..ed75692df 100644 --- a/go.sum +++ b/go.sum @@ -113,6 +113,8 @@ github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7a github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keLg81eXfW3O+oY= +github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I= github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0/go.mod h1:E/TSTwGwJL78qG/PmXZO1EjYhfJinVAhrmmHX6Z8B9k= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= @@ -172,6 +174,8 @@ github.com/gorilla/handlers v1.3.0 h1:tsg9qP3mjt1h4Roxp+M1paRjrVBfPSOpBuVclh6Ylu github.com/gorilla/handlers v1.3.0/go.mod h1:Qkdc/uu4tH4g6mTK6auzZ766c4CA0Ng8+o/OAirnOIQ= github.com/gorilla/mux v1.7.0 h1:tOSd0UKHQd6urX6ApfOn4XdBMY6Sh1MfxV3kmaazO+U= github.com/gorilla/mux v1.7.0/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs= +github.com/gorilla/securecookie v1.1.1 h1:miw7JPhV+b/lAHSXz4qd/nN9jRiAFV5FwjeKyCS8BvQ= +github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4= github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ= github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc= github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= @@ -459,6 +463,7 @@ golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d h1:20cMwl2fHAzkJMEA+8J4JgqBQ golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= +golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45 h1:SVwTIAaPC2U/AvvLNZ2a7OVsmBpC8L5BlwK1whH3hm0= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -551,6 +556,7 @@ google.golang.org/api v0.13.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsb google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM= google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= +google.golang.org/appengine v1.6.1 h1:QzqyMA1tlu6CgqCDUtU9V+ZKhLFT2dkJuANu5QaxI3I= google.golang.org/appengine v1.6.1/go.mod h1:i06prIuMbXzDqacNJfV5OdTW448YApPu5ww/cMBSeb0= google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/genproto v0.0.0-20190307195333-5fe7a883aa19/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE= diff --git a/http/handler.go b/http/handler.go index 39c99e5d2..6d7796087 100644 --- a/http/handler.go +++ b/http/handler.go @@ -29,6 +29,7 @@ import ( "github.com/gorilla/handlers" "github.com/gorilla/mux" pilosa "github.com/molecula/featurebase/v2" + "github.com/molecula/featurebase/v2/authn" "github.com/molecula/featurebase/v2/encoding/proto" "github.com/molecula/featurebase/v2/ingest" "github.com/molecula/featurebase/v2/logger" @@ -69,6 +70,8 @@ type Handler struct { middleware []func(http.Handler) http.Handler pprofCPUProfileBuffer *bytes.Buffer + + auth *authn.Auth } // externalPrefixFlag denotes endpoints that are intended to be exposed to clients. @@ -115,6 +118,13 @@ func OptHandlerAPI(api *pilosa.API) handlerOption { } } +func OptHandlerAuth(auth *authn.Auth) handlerOption { + return func(h *Handler) error { + h.auth = auth + return nil + } +} + func OptHandlerFileSystem(fs pilosa.FileSystem) handlerOption { return func(h *Handler) error { h.fileSystem = fs @@ -362,7 +372,7 @@ func (h *Handler) collectStats(next http.Handler) http.Handler { // latticeRoutes lists the frontend routes that do not directly correspond to // backend routes, and require special handling. -var latticeRoutes = []string{"/tables", "/query", "/querybuilder"} // TODO somehow pull this from some metadata in the lattice directory +var latticeRoutes = []string{"/tables", "/query", "/querybuilder", "/signin"} // TODO somehow pull this from some metadata in the lattice directory // newRouter creates a new mux http router. func newRouter(handler *Handler) http.Handler { @@ -456,6 +466,12 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/cpu-profile/start", handler.handleCPUProfileStart).Methods("GET").Name("CPUProfileStart") router.HandleFunc("/cpu-profile/stop", handler.handleCPUProfileStop).Methods("GET").Name("CPUProfileStop") + router.HandleFunc("/login", handler.handleLogin).Methods("GET").Name("Login") + router.HandleFunc("/logout", handler.handleLogout).Methods("GET").Name("Logout") + router.HandleFunc("/redirect", handler.handleRedirect).Methods("GET").Name("Redirect") + router.HandleFunc("/auth", handler.handleCheckAuthentication).Methods("GET").Name("CheckAuthentication") + router.HandleFunc("/userinfo", handler.handleUserInfo).Methods("GET").Name("UserInfo") + // Endpoints to support lattice UI embedded via statik. // The messiness here reflects the fact that assets live in a nontrivial // directory structure that is controlled externally. @@ -3365,3 +3381,73 @@ func (h *Handler) handlePostRestore(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte("OK")) //nolint:errcheck } + +func (h *Handler) handleLogin(w http.ResponseWriter, r *http.Request) { + if h.auth == nil { + w.Header().Add("Content-Type", "text/plain") + w.WriteHeader(http.StatusNoContent) + w.Write([]byte("Auth Off")) //nolint:errcheck + return + } + + h.auth.Login(w, r) +} + +func (h *Handler) handleRedirect(w http.ResponseWriter, r *http.Request) { + if h.auth == nil { + w.Header().Add("Content-Type", "text/plain") + w.WriteHeader(http.StatusNoContent) + w.Write([]byte("Auth Off")) //nolint:errcheck + return + } + h.auth.Redirect(w, r) +} + +func (h *Handler) handleCheckAuthentication(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + if h.auth == nil { + w.Header().Add("Content-Type", "text/plain") + w.WriteHeader(http.StatusNoContent) + w.Write([]byte("Auth Off")) //nolint:errcheck + return + } + groups, err := h.auth.Authenticate(w, r) + if groups == nil || err != nil { + w.Header().Add("Content-Type", "text/plain") + w.WriteHeader(http.StatusForbidden) + return + } + w.Header().Add("Content-Type", "text/plain") + w.WriteHeader(http.StatusOK) + w.Write([]byte("OK")) //nolint:errcheck + +} + +func (h *Handler) handleUserInfo(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + if h.auth == nil { + w.Header().Add("Content-Type", "text/plain") + w.WriteHeader(http.StatusNoContent) + w.Write([]byte("Auth Off")) //nolint:errcheck + return + } + if err := json.NewEncoder(w).Encode(h.auth.GetUserInfo(w, r)); err != nil { + h.logger.Errorf("writing user info: %s", err) + } +} + +func (h *Handler) handleLogout(w http.ResponseWriter, r *http.Request) { + if h.auth == nil { + w.Header().Add("Content-Type", "text/plain") + w.WriteHeader(http.StatusNoContent) + w.Write([]byte("Auth Off")) //nolint:errcheck + return + } + h.auth.Logout(w, r) +} diff --git a/http/handler_internal_test.go b/http/handler_internal_test.go index e28924035..e9cfee87a 100644 --- a/http/handler_internal_test.go +++ b/http/handler_internal_test.go @@ -3,13 +3,24 @@ package http import ( "bytes" + "encoding/hex" "encoding/json" + "io/ioutil" + gohttp "net/http" + "net/http/httptest" + "net/url" + "os" "reflect" "strings" "testing" + "time" + "github.com/gorilla/securecookie" pilosa "github.com/molecula/featurebase/v2" + "github.com/molecula/featurebase/v2/authn" + "github.com/molecula/featurebase/v2/logger" "github.com/molecula/featurebase/v2/pql" + "golang.org/x/oauth2" ) // Test custom UnmarshalJSON for postIndexRequest object @@ -166,3 +177,402 @@ func TestFieldOptionValidation(t *testing.T) { } } } + +func readResponse(w *httptest.ResponseRecorder) ([]byte, error) { + res := w.Result() + defer res.Body.Close() + return ioutil.ReadAll(res.Body) +} + +func TestHandlerAuth(t *testing.T) { + type evaluate func(w *httptest.ResponseRecorder, data []byte) + type endpoint func(w gohttp.ResponseWriter, r *gohttp.Request) + var ( + ClientId = "e9088663-eb08-41d7-8f65-efb5f54bbb71" + ClientSecret = "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF" + AuthorizeURL = "https://login.microsoftonline.com/4a137d66-d161-4ae4-b1e6-07e9920874b8/oauth2/v2.0/authorize" + TokenURL = "https://login.microsoftonline.com/4a137d66-d161-4ae4-b1e6-07e9920874b8/oauth2/v2.0/token" + GroupEndpointURL = "https://graph.microsoft.com/v1.0/me/transitiveMemberOf/microsoft.graph.group?$count=true" + LogoutURL = "https://login.microsoftonline.com/common/oauth2/v2.0/logout" + Scopes = []string{"https://graph.microsoft.com/.default", "offline_access"} + HashKey = "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF" + BlockKey = "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF" + ) + + hashKey, _ := hex.DecodeString("DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF") + blockKey, _ := hex.DecodeString("DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF") + + a, err := authn.NewAuth( + logger.NewStandardLogger(os.Stdout), + "http://localhost:10101/", + Scopes, + AuthorizeURL, + TokenURL, + GroupEndpointURL, + LogoutURL, + ClientId, + ClientSecret, + HashKey, + BlockKey, + ) + if err != nil { + t.Errorf("building auth object%s", err) + } + + h := Handler{ + auth: a, + } + + hOff := Handler{} + + token := oauth2.Token{ + TokenType: "Bearer", + RefreshToken: "abcdef", + Expiry: time.Now().Add(time.Hour), + } + + expiredToken := oauth2.Token{ + TokenType: "Bearer", + RefreshToken: "abcdef", + Expiry: time.Now(), + } + + grp := authn.Group{ + UserID: "snowstorm", + GroupID: "abcd123-A", + GroupName: "Romantic Painters", + } + + validCV := authn.CookieValue{ + UserID: "snowstorm", + UserName: "J.M.W. Turner", + GroupMembership: []authn.Group{grp}, + Token: &token, + } + + emptyCV := authn.CookieValue{ + UserID: "narcissus", + UserName: "Caravaggio", + GroupMembership: []authn.Group{}, + Token: &token, + } + expiredCV := authn.CookieValue{ + UserID: "narcissus", + UserName: "Caravaggio", + GroupMembership: []authn.Group{}, + Token: &expiredToken, + } + + secure := securecookie.New(hashKey, blockKey) + validEncodedCV, _ := secure.Encode("molecula-chip", validCV) + noGroupEncodedCV, _ := secure.Encode("molecula-chip", emptyCV) + expiredEncodedCV, _ := secure.Encode("molecula-chip", expiredCV) + + validCookie := &gohttp.Cookie{ + Name: "molecula-chip", + Value: validEncodedCV, + Path: "/", + Secure: true, + HttpOnly: true, + Expires: token.Expiry, + } + noGroupCookie := &gohttp.Cookie{ + Name: "molecula-chip", + Value: noGroupEncodedCV, + Path: "/", + Secure: true, + HttpOnly: true, + Expires: token.Expiry, + } + expiredCookie := &gohttp.Cookie{ + Name: "molecula-chip", + Value: expiredEncodedCV, + Path: "/", + Secure: true, + HttpOnly: true, + Expires: time.Now().Add(time.Minute * -1), + } + emptyCookie := &gohttp.Cookie{ + Name: "molecula-chip", + Value: "", + Path: "/", + Secure: true, + HttpOnly: true, + Expires: token.Expiry, + } + unEncodedCookie := &gohttp.Cookie{ + Name: "molecula-chip", + Value: "The quick brown fox", + Path: "/", + Secure: true, + HttpOnly: true, + Expires: token.Expiry, + } + + tests := []struct { + name string + path string + kind string + cookie *gohttp.Cookie + handler endpoint + fn evaluate + }{ + { + name: "Login", + path: "/login", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleLogin(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if strings.Index(string(data), AuthorizeURL) != 9 { + t.Errorf("incorrect redirect url: expected: %s, got: %s", AuthorizeURL, string(data)) + } + }, + }, + { + name: "Logout", + path: "/logout", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleLogout(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if w.Result().Cookies()[0].Value != "" { + t.Errorf("expected cookie to be cleared, got: %+v", w.Result().Cookies()[0].Value) + } + }, + }, + { + name: "Authenticate-Groups", + path: "/auth", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleCheckAuthentication(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if w.Result().StatusCode != 200 { + t.Errorf("expected http code 200, got: %+v", w.Result().StatusCode) + } + }, + }, + { + name: "Authenticate-NoGroups", + path: "/auth", + kind: "type1", + cookie: noGroupCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleCheckAuthentication(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + // status forbidden + if w.Result().StatusCode != 403 { + t.Errorf("expected http code 403, got: %+v", w.Result().StatusCode) + } + }, + }, + { + name: "Authenticate-MalformedCookie", + path: "/auth", + kind: "type1", + cookie: unEncodedCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleCheckAuthentication(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + // redirect to signin + if w.Result().StatusCode != 307 { + t.Errorf("expected http code 307, got: %+v", w.Result().StatusCode) + } + }, + }, + { + name: "Authenticate-Expired", + path: "/auth", + kind: "type1", + cookie: expiredCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleCheckAuthentication(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + // redirect to signin + if w.Result().StatusCode != 307 { + t.Errorf("expected http code 307, got: %+v", w.Result().StatusCode) + } + }, + }, + { + name: "Authenticate-NoCookie", + path: "/auth", + kind: "type1", + cookie: emptyCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleCheckAuthentication(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + // redirect to signin + if w.Result().StatusCode != 307 { + t.Errorf("expected http code 307, got: %+v", w.Result().StatusCode) + } + }, + }, + { + name: "UserInfo", + path: "/userinfo", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleUserInfo(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + uinfo := authn.UserInfo{} + err = json.Unmarshal(data, &uinfo) + if err != nil { + t.Errorf("unmarshalling userinfo") + } + if uinfo.UserID != "snowstorm" && uinfo.UserName != "J.M.W. Turner" { + t.Errorf("expected http code 400, got: %+v", uinfo) + } + }, + }, + { + name: "UserInfo-NoCookie", + path: "/userinfo", + kind: "type1", + cookie: emptyCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleUserInfo(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + uinfo := authn.UserInfo{} + err = json.Unmarshal(data, &uinfo) + if err != nil { + t.Errorf("unmarshalling userinfo") + } + if uinfo.UserID != "" && uinfo.UserName != "" { + t.Errorf("expected http code 400, got: %+v", uinfo) + } + }, + }, + + { + name: "Redirect-NoAuthCode", + path: "/redirect", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleRedirect(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if strings.Index(string(data), AuthorizeURL) != 9 { + if w.Result().StatusCode != 400 { + t.Errorf("expected http code 400, got: %+v", w.Result().StatusCode) + } + } + }, + }, + { + name: "Redirect-SomeAuthCode", + path: "/redirect", + kind: "type2", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { h.handleRedirect(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if strings.Index(string(data), AuthorizeURL) != 9 { + if w.Result().StatusCode != 400 { + t.Errorf("expected http code 400, got: %+v", w.Result().StatusCode) + } + } + }, + }, + { + name: "Login-AuthOff", + path: "/login", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { hOff.handleLogin(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if strings.Index(string(data), AuthorizeURL) != 9 { + if w.Result().StatusCode != 204 { + t.Errorf("expected http code 204, got: %+v", w.Result().StatusCode) + } + } + }, + }, + { + name: "Logout-AuthOff", + path: "/logout", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { hOff.handleLogout(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if strings.Index(string(data), AuthorizeURL) != 9 { + if w.Result().StatusCode != 204 { + t.Errorf("expected http code 204, got: %+v", w.Result().StatusCode) + } + } + }, + }, + { + name: "UserInfo-AuthOff", + path: "/userinfo", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { hOff.handleUserInfo(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if strings.Index(string(data), AuthorizeURL) != 9 { + if w.Result().StatusCode != 204 { + t.Errorf("expected http code 204, got: %+v", w.Result().StatusCode) + } + } + }, + }, + { + name: "Authenticate-AuthOff", + path: "/auth", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { hOff.handleCheckAuthentication(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if strings.Index(string(data), AuthorizeURL) != 9 { + if w.Result().StatusCode != 204 { + t.Errorf("expected http code 204, got: %+v", w.Result().StatusCode) + } + } + }, + }, + { + name: "Redirect-AuthOff", + path: "/redirect", + kind: "type1", + cookie: validCookie, + handler: func(w gohttp.ResponseWriter, r *gohttp.Request) { hOff.handleRedirect(w, r) }, + fn: func(w *httptest.ResponseRecorder, data []byte) { + if strings.Index(string(data), AuthorizeURL) != 9 { + if w.Result().StatusCode != 204 { + t.Errorf("expected http code 204, got: %+v", w.Result().StatusCode) + } + } + }, + }, + } + + for _, test := range tests { + switch test.kind { + case "type1": + t.Run(test.name, func(t *testing.T) { + r := httptest.NewRequest(gohttp.MethodGet, test.path, nil) + w := httptest.NewRecorder() + r.AddCookie(test.cookie) + test.handler(w, r) + data, err := readResponse(w) + if err != nil { + t.Errorf("expected no errors reading response, got: %+v", err) + } + test.fn(w, data) + }) + case "type2": + t.Run(test.name, func(t *testing.T) { + r := httptest.NewRequest(gohttp.MethodGet, test.path, nil) + w := httptest.NewRecorder() + r.Form = url.Values{} + r.Header.Set("Content-Type", "application/x-www-form-urlencoded") + r.Form.Add("code", "junk") + + test.handler(w, r) + data, err := readResponse(w) + if err != nil { + t.Errorf("expected no errors reading response, got: %+v", err) + } + + test.fn(w, data) + + }) + } + + } + +} diff --git a/install/featurebase.conf b/install/featurebase.conf index 540a410f4..033db191d 100644 --- a/install/featurebase.conf +++ b/install/featurebase.conf @@ -372,7 +372,8 @@ log-path = "/var/log/molecula/featurebase.log" # ============================================================================== # Enable/Disable AuthN/AuthZ for featurebase -# Can choose identity provider, pass authorize and user-info endpoints, and client id +# Can choose identity provider, defaults for Azure Active Directory +# Use provided keygen binary to generate hash and block keys with sufficient length and entropy # [auth] # enable = false # client-id = "" @@ -380,4 +381,8 @@ log-path = "/var/log/molecula/featurebase.log" # authorize-url = "" # token-url = "" # group-endpoint-url = "" -# scope-url = "" \ No newline at end of file +# logout-url = "" +# scopes = ["", ""] +# hash-key = "" +# block-key = "" +# permissions = "" diff --git a/lattice/src/App.tsx b/lattice/src/App.tsx index f465bd480..4a93cd6ff 100644 --- a/lattice/src/App.tsx +++ b/lattice/src/App.tsx @@ -1,58 +1,37 @@ -import React, { useEffect, useState } from 'react'; -import CssBaseline from '@material-ui/core/CssBaseline'; -import { Route, Switch } from 'react-router-dom'; -import { darkTheme, lightTheme } from 'theme/'; -import { Home } from 'App/Home'; -import { Header } from 'shared/Header'; +import { BrowserRouter, Route, Switch } from 'react-router-dom'; import { MuiThemeProvider } from '@material-ui/core/styles'; -import { Nav } from 'shared/Nav'; -import { NotFound } from 'App/NotFound'; -import { MoleculaTablesContainer } from 'App/MoleculaTables'; -import { QueryContainer } from 'App/Query'; -import { QueryBuilderContainer } from 'App/QueryBuilder'; -import css from './App.module.scss'; + +import { useAuth } from 'services/useAuth'; +import PrivateRoute from 'shared/PrivateRoute/PrivateRoute'; +import { lightTheme } from 'theme/'; +import Main from 'Main'; +import Signin from 'App/AuthFlow/Signin'; const App = () => { - const [theme, setTheme] = useState( - localStorage.getItem('theme') || 'light' - ); - - useEffect(() => { - if(theme === 'dark') { - document.documentElement.setAttribute('data-theme', 'dark') - } else { - document.documentElement.removeAttribute('data-theme'); - } - }, [theme]); - - const onToggleTheme = () => { - const newTheme = theme === 'dark' ? 'light' : 'dark'; - setTheme(newTheme); - localStorage.setItem('theme', newTheme); - }; + const auth = useAuth(); return ( - - -
- -
-
-
-
- + ) : ( + // Auth is off, all routes are accessible + + )} + + )} + ); -} +}; export default App; diff --git a/lattice/src/App/AuthFlow/AuthFlow.module.scss b/lattice/src/App/AuthFlow/AuthFlow.module.scss new file mode 100644 index 000000000..5ac275622 --- /dev/null +++ b/lattice/src/App/AuthFlow/AuthFlow.module.scss @@ -0,0 +1,56 @@ +.main { + min-height: 100vh; + background-repeat: no-repeat; + background-image: linear-gradient( + to bottom, + rgba(250, 250, 250, 1), + rgba(250, 250, 250, 0.7) + ), + url(/assets/bg-pattern.png); + background-size: cover; + padding-bottom: 32px; +} + +.logoContainer { + text-align: center; +} + +.logo { + height: 85px; + margin: 16px; +} + +.loginForm { + width: 500px; + margin: 0 auto; + padding-top: 75px; +} + +.formError { + color: #f44336; + margin-bottom: 16px; +} + +.sso { + text-align: center; + padding: 24px 0 16px; +} + +.passwordField { + position: relative; + + .forgotPassword { + // [syang] Eww yes, I hate this + position: absolute; + right: 0; + z-index: 1; + } +} + +.backToSignIn { + padding: 24px 0 16px; +} + +.alert { + margin-bottom: 16px; +} diff --git a/lattice/src/App/AuthFlow/SignInButton.tsx b/lattice/src/App/AuthFlow/SignInButton.tsx new file mode 100644 index 000000000..7ebd9b8b7 --- /dev/null +++ b/lattice/src/App/AuthFlow/SignInButton.tsx @@ -0,0 +1,20 @@ +import React from 'react'; +import { Button } from '@material-ui/core'; + +interface Props { + children?: React.ReactNode; +} + +const SignInButton: React.FC = ({ children }) => { + const signinOnClick = (e) => { + window.location.href = '/login'; + }; + + return ( + + ); +}; + +export default SignInButton; diff --git a/lattice/src/App/AuthFlow/SignOutButton.tsx b/lattice/src/App/AuthFlow/SignOutButton.tsx new file mode 100644 index 000000000..3e76c22db --- /dev/null +++ b/lattice/src/App/AuthFlow/SignOutButton.tsx @@ -0,0 +1,20 @@ +import React from 'react'; +import { Button } from '@material-ui/core'; + +interface Props { + children?: React.ReactNode; +} + +const SignOutButton: React.FC = ({ children }) => { + const signoutOnClick = (e) => { + window.location.href = '/logout'; + }; + + return ( + + ); +}; + +export default SignOutButton; diff --git a/lattice/src/App/AuthFlow/Signin.tsx b/lattice/src/App/AuthFlow/Signin.tsx new file mode 100644 index 000000000..4a5bb8ce2 --- /dev/null +++ b/lattice/src/App/AuthFlow/Signin.tsx @@ -0,0 +1,30 @@ +import Card from '@material-ui/core/Card'; +import CardContent from '@material-ui/core/CardContent'; +import CardHeader from '@material-ui/core/CardHeader'; + +import { ReactComponent as MLogo } from 'assets/m-bug-alt.svg'; +import css from './AuthFlow.module.scss'; +import SignInButton from './SignInButton'; + +function Signin(props) { + const renderLoginForm = () => ( + + + + + + + ); + + return ( +
+
+
+ +
+ {renderLoginForm()} +
+
+ ); +} +export default Signin; diff --git a/lattice/src/App/AuthFlow/index.ts b/lattice/src/App/AuthFlow/index.ts new file mode 100644 index 000000000..364a48925 --- /dev/null +++ b/lattice/src/App/AuthFlow/index.ts @@ -0,0 +1 @@ +export * from './Signin'; \ No newline at end of file diff --git a/lattice/src/Main.tsx b/lattice/src/Main.tsx new file mode 100644 index 000000000..b29410889 --- /dev/null +++ b/lattice/src/Main.tsx @@ -0,0 +1,58 @@ +import { useEffect, useState } from 'react'; +import { Route, Switch } from 'react-router-dom'; +import CssBaseline from '@material-ui/core/CssBaseline'; +import { MuiThemeProvider } from '@material-ui/core/styles'; + +import { Header } from 'shared/Header'; +import { Nav } from 'shared/Nav'; +import { darkTheme, lightTheme } from 'theme/'; +import { Home } from 'App/Home'; +import { MoleculaTablesContainer } from 'App/MoleculaTables'; +import { NotFound } from 'App/NotFound'; +import { QueryContainer } from 'App/Query'; +import { QueryBuilderContainer } from 'App/QueryBuilder'; + +import css from './App.module.scss'; + +const Main = () => { + const [theme, setTheme] = useState(localStorage.getItem('theme') || 'light'); + + useEffect(() => { + if (theme === 'dark') { + document.documentElement.setAttribute('data-theme', 'dark'); + } else { + document.documentElement.removeAttribute('data-theme'); + } + }, [theme]); + + const onToggleTheme = () => { + const newTheme = theme === 'dark' ? 'light' : 'dark'; + setTheme(newTheme); + localStorage.setItem('theme', newTheme); + }; + + return ( +
+ + +
+
+
+
+
+ +
+ ); +}; + +export default Main; diff --git a/lattice/src/assets/bg-pattern.png b/lattice/src/assets/bg-pattern.png new file mode 100644 index 000000000..23bdf09be Binary files /dev/null and b/lattice/src/assets/bg-pattern.png differ diff --git a/lattice/src/assets/m-bug-alt.svg b/lattice/src/assets/m-bug-alt.svg new file mode 100644 index 000000000..a0cc81bc1 --- /dev/null +++ b/lattice/src/assets/m-bug-alt.svg @@ -0,0 +1,16 @@ + + + + + + + + + + + + + + + + diff --git a/lattice/src/index.tsx b/lattice/src/index.tsx index 331a583a3..dccfe3cab 100644 --- a/lattice/src/index.tsx +++ b/lattice/src/index.tsx @@ -1,14 +1,17 @@ import React from 'react'; import ReactDOM from 'react-dom'; -import App from './App'; -import { BrowserRouter as Router, Route } from 'react-router-dom'; +import { ProvideAuth } from 'services/useAuth'; + import * as serviceWorker from './serviceWorker'; import './index.scss'; +import App from './App'; ReactDOM.render( - - - , + + + + + , document.getElementById('root') ); diff --git a/lattice/src/services/eventServices.tsx b/lattice/src/services/eventServices.tsx index 5ad7e19db..b2adcfd33 100644 --- a/lattice/src/services/eventServices.tsx +++ b/lattice/src/services/eventServices.tsx @@ -1,12 +1,13 @@ import axios from 'axios'; + import { baseURL } from './baseURL'; const api = axios.create({ baseURL, headers: { 'Content-Type': 'application/x-www-form-urlencoded', - Accept: 'application/json' - } + Accept: 'application/json', + }, }); export const pilosa = { @@ -14,6 +15,12 @@ export const pilosa = { status() { return api.get('/status'); }, + auth() { + return api.get('/auth'); + }, + userinfo() { + return api.get('/userinfo'); + }, info() { return api.get('/info'); }, @@ -40,7 +47,7 @@ export const pilosa = { }, queryHistory() { return api.get('/query-history'); - } + }, }, post: { finishTransaction(id) { @@ -48,6 +55,6 @@ export const pilosa = { }, query(index, query) { return api.post(`/index/${index}/query`, query); - } - } + }, + }, }; diff --git a/lattice/src/services/useAuth.tsx b/lattice/src/services/useAuth.tsx new file mode 100644 index 000000000..0c94cb733 --- /dev/null +++ b/lattice/src/services/useAuth.tsx @@ -0,0 +1,81 @@ +import React, { createContext, useContext, useEffect, useState } from 'react'; + +import { pilosa } from './eventServices'; + +const authContext = createContext({}); + +// Provider component that wraps your app and makes auth object ... +// ... available to any child component that calls useAuth(). +export function ProvideAuth({ children }) { + const auth = useProvideAuth(); + return {children}; +} + +// Hook for child components to get the auth object ... +// ... and re-render when it changes. +export const useAuth = () => { + return useContext(authContext); +}; + +export interface IUser { + userid: string; + username: string; +} + +// Provider hook that creates auth object and handles state +function useProvideAuth() { + const [user, setUser] = useState(undefined); + const [isAuthenticated, setIsAuthenticated] = useState(false); + const [isLoading, setIsLoading] = useState(true); + const [isAuthOn, setIsAuthOn] = useState(true); + + const userinfo = () => { + pilosa.get.userinfo().then((userinfoRes) => { + if (userinfoRes.data.userid && userinfoRes.data.username) { + setUser(userinfoRes.data); + } else { + setUser(undefined); + } + }); + }; + + // Subscribe to user on mount + // Because this sets state in the callback it will cause any ... + // ... component that utilizes this hook to re-render with the ... + // ... latest auth object. + useEffect(() => { + pilosa.get + .auth() + .then((res) => { + if (res.status === 204) { + // Authentication is off + setIsAuthOn(false); + } else { + // Turn on Authentication + setIsAuthOn(true); + + if (res.data === 'OK') { + // User is authenticated + setIsAuthenticated(true); + + // get userinfo + userinfo(); + } else { + // User not authenticated + setIsAuthenticated(false); + } + } + }) + .finally(() => { + setIsLoading(false); + }); + }, []); + + return { + isAuthenticated, + isLoading, + isAuthOn, + user, + userinfo, + }; +} diff --git a/lattice/src/shared/Header/Header.tsx b/lattice/src/shared/Header/Header.tsx index ed2c51bc4..b32e72e63 100644 --- a/lattice/src/shared/Header/Header.tsx +++ b/lattice/src/shared/Header/Header.tsx @@ -1,12 +1,17 @@ -import React, { FC } from 'react'; -import AppBar from '@material-ui/core/AppBar'; -import Toolbar from '@material-ui/core/Toolbar'; -import { Link } from 'react-router-dom'; -import { ReactComponent as MoleculaLogo } from 'assets/lightTheme/MoleculaLogo.svg'; -import { ReactComponent as MoleculaLogoDark } from 'assets/darkTheme/MoleculaLogo.svg'; -import { ThemeToggle } from 'shared/ThemeToggle'; -import { useTheme } from '@material-ui/core/styles'; -import css from './Header.module.scss'; +import SignOutButton from "App/AuthFlow/SignOutButton"; +import { ReactComponent as MoleculaLogoDark } from "assets/darkTheme/MoleculaLogo.svg"; +import { ReactComponent as MoleculaLogo } from "assets/lightTheme/MoleculaLogo.svg"; +import { FC } from "react"; +import { Link } from "react-router-dom"; +import { useAuth } from "services/useAuth"; +import { ThemeToggle } from "shared/ThemeToggle"; + +import AppBar from "@material-ui/core/AppBar"; +import Button from "@material-ui/core/Button"; +import { useTheme } from "@material-ui/core/styles"; +import Toolbar from "@material-ui/core/Toolbar"; + +import css from "./Header.module.scss"; type HeaderProps = { onToggleTheme: () => void; @@ -14,7 +19,8 @@ type HeaderProps = { export const Header: FC = ({ onToggleTheme }) => { const theme = useTheme(); - const isDark = theme.palette.type === 'dark'; + const isDark = theme.palette.type === "dark"; + const auth = useAuth(); return ( = ({ onToggleTheme }) => { /> + + {auth.isAuthenticated ? ( +
+ {auth.user && ( + + )} + +
+ ) : null} diff --git a/lattice/src/shared/PrivateRoute/PrivateRoute.tsx b/lattice/src/shared/PrivateRoute/PrivateRoute.tsx new file mode 100644 index 000000000..5a76677fc --- /dev/null +++ b/lattice/src/shared/PrivateRoute/PrivateRoute.tsx @@ -0,0 +1,33 @@ +import { Redirect, Route } from 'react-router-dom'; + +import { useAuth } from 'services/useAuth'; + +function PrivateRoute({ component: Component, ...rest }) { + const auth = useAuth(); + + return ( + { + if (auth.isAuthenticated) { + // If the user is authenticated, render the component + return ; + } else { + // If the user is not authenticated, redirect to sign in page + return ( + + ); + } + }} + /> + ); +} + +export default PrivateRoute; diff --git a/rbf/cfg/cfg.go b/rbf/cfg/cfg.go index cc2cf7a8a..671c6fe43 100644 --- a/rbf/cfg/cfg.go +++ b/rbf/cfg/cfg.go @@ -2,6 +2,7 @@ package cfg import ( + "github.com/molecula/featurebase/v2/logger" "github.com/spf13/pflag" ) @@ -35,6 +36,11 @@ type Config struct { // CursorCacheSize is the number of copies of Cursor{} to keep in our // readyCursorCh arena to avoid GC pressure. CursorCacheSize int64 `toml:"cursor-cache-size"` + + // Logger specifies a logger for asynchronous errors, such as + // background checkpoints. It cannot be set from toml. The default is + // to use stderr. + Logger logger.Logger `toml:"-"` } func NewDefaultConfig() *Config { diff --git a/rbf/cursor.go b/rbf/cursor.go index 68c165d94..eda6bec1d 100644 --- a/rbf/cursor.go +++ b/rbf/cursor.go @@ -774,6 +774,25 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) { cells[len(cells)-1] = branchCell{} cells = cells[:len(cells)-1] + // Branches are not allowed to have zero element so we must remove the page + // or, in the case of the root page, convert to a leaf page. + if len(cells) == 0 { + // If this is the root page, convert to leaf page. + if stackIndex == 0 { + var buf [PageSize]byte + writePageNo(buf[:], elem.pgno) + writeFlags(buf[:], PageTypeLeaf) + writeCellN(buf[:], len(cells)) + return c.tx.writePage(buf[:]) + } + + // If this is a non-root page, free and remove from parent. + if err := c.tx.freePgno(elem.pgno); err != nil { + return err + } + return c.deleteBranchCell(stackIndex-1, oldPageKey) + } + // If the root only has one node, replace it with its child. if stackIndex == 0 && len(cells) == 1 { target, _, err := c.tx.readPage(cells[0].ChildPgno) @@ -802,6 +821,9 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) { writeBranchCell(buf[:], j, offset, cell) offset += align8(branchCellSize) } + + assert(readCellN(buf[:]) > 0) // must have at least one cell + if err := c.tx.writePage(buf[:]); err != nil { return err } diff --git a/rbf/cursor_test.go b/rbf/cursor_test.go index 4f7464bd1..940798c04 100644 --- a/rbf/cursor_test.go +++ b/rbf/cursor_test.go @@ -973,8 +973,8 @@ func TestCursor_SplitBranchCells(t *testing.T) { } // c, _ := tx.Cursor("x") //added just for dot code coverage - c.Dump("ignore for coverage") - + c.Dump("test.dump") + os.Remove("test.dump") } func TestCursor_RemoveCells(t *testing.T) { diff --git a/rbf/db.go b/rbf/db.go index 99c4f0e2f..65dd4fbea 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -11,6 +11,7 @@ import ( "syscall" "github.com/benbjohnson/immutable" + "github.com/molecula/featurebase/v2/logger" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "github.com/molecula/featurebase/v2/syswrap" ) @@ -27,6 +28,16 @@ var cursorSyncPool = &sync.Pool{ }, } +// txWaiter is a representation of "i need to wait for txs to complete". +// it is created with a function, and will run that function, with the db +// lock held, at some point after every Tx that was open when it was created +// has closed. WARNING: A txWaiter may hold db.rwmu. +type txWaiter struct { + ready chan struct{} + waitingOn map[*Tx]struct{} + callback func() +} + // DB options like MaxSize, FsyncEnabled, DoAllocZero // can be set before calling DB.Open(). type DB struct { @@ -38,15 +49,21 @@ type DB struct { pageMap *PageMap // pgno-to-WALID mapping txs map[*Tx]struct{} // active transactions opened bool // true if open + logger logger.Logger // for diagnostics from async things - wal []byte // wal mmap - walFile *os.File // wal file descriptor - walPageN int // wal page count + wal []byte // wal mmap + walFile *os.File // wal file descriptor + walPageN int // wal page count + baseWALID int64 // WAL ID of first page mu sync.RWMutex // general mutex rwmu sync.Mutex // mutex for restricting single writer haltCond *sync.Cond // condition for resuming txs after checkpoint + txWaiters []*txWaiter // things waiting for Txs to close + + isDead error // this database died in an unrecoverable way, error out opens + // Path represents the path to the database file. Path string } @@ -62,6 +79,11 @@ func NewDB(path string, cfg *rbfcfg.Config) *DB { txs: make(map[*Tx]struct{}), pageMap: NewPageMap(), Path: path, + logger: cfg.Logger, + } + if db.logger == nil { + // default to writing to stdout if not told otherwise + db.logger = logger.NewStandardLogger(os.Stderr) } db.haltCond = sync.NewCond(&db.mu) @@ -133,8 +155,12 @@ func (db *DB) Open() (err error) { // Open write-ahead log & checkpoint to the end since no transactions are open. if err := db.openWAL(); err != nil { return fmt.Errorf("wal open: %w", err) - } else if err := db.checkpoint(); err != nil { - return fmt.Errorf("checkpoint: %w", err) + } else { + // checkpoint wants to hold the rwmu lock. + db.rwmu.Lock() + if err := db.checkpoint(); err != nil { + return fmt.Errorf("startup checkpoint: %w", err) + } } return nil @@ -158,10 +184,12 @@ func (db *DB) openWAL() (err error) { // Determine the number of whole pages in the WAL. var pageN int + var fileSize int64 if fi, err := db.walFile.Stat(); err != nil { return fmt.Errorf("wal stat: %w", err) } else { - pageN = int(fi.Size() / PageSize) + fileSize = fi.Size() + pageN = int(fileSize / PageSize) } // Read backwards through the WAL to find the last valid meta page. @@ -169,28 +197,96 @@ func (db *DB) openWAL() (err error) { if page, err := db.readWALPageAt(pageN - 1); err != nil { return err } else if IsMetaPage(page) { + // We now face a challenge. Probably this is a meta page. + // But consider a sequence of pages written which gets + // interrupted right before the meta page is written. + // If the last page is a bitmap page, it could LOOK LIKE a meta + // page. So we have to check the page before it. If that page + // is a bitmap header, then actually this is a bitmap page, right? + // If that page doesn't exist, of course, we're fine, except + // for the philosophical question of why we wrote a meta page + // when no pages had changed. + if pageN > 1 { + if page, err = db.readWALPageAt(pageN - 2); err != nil { + return err + } + if IsBitmapHeader(page) { + // But wait! + // What if this *is* a meta page, and the page before it is + // actually a *bitmap page* that looks like a bitmap header? And + // so on. + // + // Rather than try to resolve this, in this insanely unlikely + // situation, we read from the beginning which allows us to + // always know what we're seeing, because every bitmap page + // comes *after* a bitmap header page, and thus, we know when + // we might be seeing one. + pageN, err = db.methodicalWALPageN(pageN) + if err != nil { + return err + } + } + } break } } - - // Truncate WAL to the last valid meta page. - if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil { - return fmt.Errorf("wal truncate: %w", err) - } else if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil { + if fileSize != int64(pageN*PageSize) { + if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil { + return fmt.Errorf("wal truncate: %w", err) + } + } + if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil { return fmt.Errorf("wal seek: %w", err) } db.walPageN = pageN + db.baseWALID = readMetaWALID(db.data) return nil } -// checkpoint moves all WAL pages to the main DB file. -// Must be called by a write transaction while under db.mu lock. -func (db *DB) checkpoint() error { +// methodicalWALPageN tries to determine the last meta page in a very reliable +// but slow way. This handles the theoretical but hard to imagine creating +// edge case where we have a bitmap page which happens to look like a meta +// page, and the write got interrupted before the meta page got written. +func (db *DB) methodicalWALPageN(pageN int) (lastMeta int, err error) { + for i := 0; i < pageN; i++ { + var page []byte + if page, err = db.readWALPageAt(i); err != nil { + return -1, err + } + switch { + case IsMetaPage(page): + lastMeta = i + case IsBitmapHeader(page): + // skip the bitmap page, which we can't usefully evaluate + i++ + } + } + return lastMeta, nil +} + +// Checkpoint performs a manual checkpoint. This is not necessary except for tests. +func (db *DB) Checkpoint() error { + db.mu.Lock() + defer db.mu.Unlock() + db.rwmu.Lock() + return db.checkpoint() +} + +// checkpoint moves all WAL pages to the main DB file. Must be called +// while holding both db.mu and db.rwmu. Should release db.rwmu, but not +// db.mu. +func (db *DB) checkpoint() (err error) { + // if we don't spin off a possible async waiter, we should release the + // write lock, if we do, that will release it. + releaseLock := true + defer func() { + if releaseLock { + db.rwmu.Unlock() + } + }() if !db.opened { return nil - } else if len(db.txs) > 0 { - return nil // skip if transactions open } // Check if there are any WAL pages, if not do nothing as @@ -199,48 +295,112 @@ func (db *DB) checkpoint() error { if db.walPageN == 0 { return nil } - - for i := 0; i < db.walPageN; i++ { - page, err := db.readWALPageAt(i) - if err != nil { - return err + // wake up things waiting on haltCond when we're done, even if we fail. + // Otherwise, we deadlock with them all stuck waiting on that forever. + defer func() { + if err != nil && db.isDead == nil { + db.isDead = err } + db.haltCond.Broadcast() + }() - // Determine page number. Meta pages are always on zero & bitmap - // headers specify the page number of the next page in the WAL. - // All other pages have their page number in the page data. - var pgno uint32 - if IsBitmapHeader(page) { - pgno = readPageNo(page) - if page, err = db.readWALPageAt(i + 1); err != nil { - return err + // Copy the pages from the WAL back to the database outside of the lock. + if err := func() error { + db.mu.Unlock() // This is intentionally reversed so run w/o lock + defer db.mu.Lock() + + var page []byte + // We might have either a *PageMap or just the file. If we have the file, + // building the PageMap is fairly expensive because it's fancy and immutable. + // If we have the PageMap *or* some other map, that's two different things + // to iterate. If we have the PageMap, building a map from it is relatively + // cheap, so we'll do it that way. + pages := make(map[uint32]int) + + if db.pageMap.size == 0 { + // you'd think we're done, but actually this PROBABLY means that + // this is initial startup, and we haven't read the file yet. We scan + // the file for pages, because it turns out most of them probably + // got overwritten. + for i := 0; i < db.walPageN; i++ { + page, err = db.readWALPageAt(i) + if err != nil { + return fmt.Errorf("reading WAL page %d: %w", i, err) + } + + // Determine page number. Meta pages are always on zero & bitmap + // headers specify the page number of the next page in the WAL. + // All other pages have their page number in the page data. + var pgno uint32 + if IsBitmapHeader(page) { + pgno = readPageNo(page) + if i+1 < db.walPageN { + if page, err = db.readWALPageAt(i + 1); err != nil { + return err + } + } else { + return fmt.Errorf("last page of WAL file (%d) is bitmap header", i) + } + i++ // bitmaps in WAL are two pages + } else if !IsMetaPage(page) { + pgno = readPageNo(page) + } + // record where in the file we have this page + pages[pgno] = i + } + } else { + itr := db.pageMap.Iterator() + itr.First() + for k, v, ok := itr.Next(); ok; k, v, ok = itr.Next() { + pages[k] = int(v - db.baseWALID - 1) } - i++ // bitmaps in WAL are two pages - } else if !IsMetaPage(page) { - pgno = readPageNo(page) } - // Write data to the data file. - if err := db.writeDBPage(pgno, page); err != nil { - return err + // fmt.Printf("checkpoint: walPageN %d, PageMap size %d\n", db.walPageN, db.pageMap.size) + for pgno, walID := range pages { + page, err = db.readWALPageAt(walID) + if err != nil { + return fmt.Errorf("reading page %d [page number %d]: %v", walID, pgno, err) + } + + // Write data to the data file. + if err = db.writeDBPage(pgno, page); err != nil { + return fmt.Errorf("writing page %d: %v", pgno, err) + } } + + // Ensure database file is synced and then truncate the WAL file. + if err = db.fsync(db.file); err != nil { + return fmt.Errorf("db file sync: %w", err) + } + + return nil + }(); err != nil { + return err } - // Ensure database file is synced and then truncate the WAL file. - if err := db.fsync(db.file); err != nil { - return fmt.Errorf("db file sync: %w", err) - } else if err := db.walFile.Truncate(0); err != nil { - return fmt.Errorf("truncate wal file: %w", err) - } else if err := db.fsync(db.walFile); err != nil { - return fmt.Errorf("wal file sync: %w", err) - } else if _, err := db.walFile.Seek(0, io.SeekStart); err != nil { - return fmt.Errorf("seek wal file: %w", err) - } + // now we've updated the file. There are existing transactions that are still + // using the WAL, though. So we wait for them to terminate before we unlock + // the rwmu and update the metadata about the WAL. + releaseLock = false db.walPageN = 0 db.pageMap = NewPageMap() - // Notify halted transactions that the WAL has been checkpointed. - db.haltCond.Broadcast() + db.afterCurrentTx(func() { + defer db.rwmu.Unlock() + db.baseWALID = readMetaWALID(db.data) + db.mu.Unlock() + defer db.mu.Lock() + + if err = db.walFile.Truncate(0); err != nil { + db.logger.Errorf("truncate wal file: %w", err) + } else if err = db.fsync(db.walFile); err != nil { + db.logger.Errorf("wal file sync: %w", err) + } else if _, err = db.walFile.Seek(0, io.SeekStart); err != nil { + db.logger.Errorf("seek wal file: %w", err) + } + + }) return nil } @@ -450,10 +610,25 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { cleanup() return nil, ErrClosed } + if db.isDead != nil { + err := db.isDead + cleanup() + return nil, err + } - // Wait for WAL size to be below threshold. - for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize { - db.haltCond.Wait() + // Wait for WAL size to be below threshold, if we're going to write. + // Reads don't care. + if writable { + for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize { + if db.isDead != nil { + err := db.isDead + cleanup() + return nil, err + } + // This implicitly releases db.mu.Lock and comes back with it + // held again. + db.haltCond.Wait() + } } tx := &Tx{ @@ -502,26 +677,95 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { return tx, nil } -// removeTx removes an active transaction from the database. -func (db *DB) removeTx(tx *Tx) error { - // Release writer lock if tx is writable. - if tx.writable { - tx.db.rwmu.Unlock() +// afterCurrentTx produces runs the provided callback, with the db lock +// held, after all current Tx terminate. It should be called with the db +// lock held. +func (db *DB) afterCurrentTx(callback func()) { + if len(db.txs) == 0 { + callback() + return } + txw := &txWaiter{} + txw.ready = make(chan struct{}) + txw.callback = callback + txw.waitingOn = make(map[*Tx]struct{}, len(db.txs)) + for k := range db.txs { + txw.waitingOn[k] = struct{}{} + } + db.txWaiters = append(db.txWaiters, txw) + go func() { + <-txw.ready + // fmt.Printf("afterCurrentTx: locking db\n") + db.mu.Lock() + defer db.mu.Unlock() + // fmt.Printf("afterCurrentTx: running callback\n") + txw.callback() + }() + return +} +// removeTx removes an active transaction from the database. it obtains +// the db lock, and currently drops it, but will later possibly be leaving +// it retained by an asynchronous op that wants to happen before we start +// running new tx. +func (db *DB) removeTx(tx *Tx) error { + // We might want to trigger a checkpoint. Only for writable + // transactions, and only when either there's nothing else open or we + // really need to. + checkpoint := false + if tx.writable { + walSize := db.walSize() + if walSize > db.cfg.MinWALCheckpointSize { + // Might be a good time for a checkpoint. We'll do a checkpoint + // if we're the only transaction, or if we have to. + if len(db.txs) == 1 || walSize > db.cfg.MaxWALCheckpointSize { + checkpoint = true + } + } + // During checkpointing, we'll be preventing writes, but allowing reads. + if !checkpoint { + tx.db.rwmu.Unlock() + } + } + // remove ourselves from the list of transactions the db is keeping. delete(tx.db.txs, tx) + for i := 0; i < len(tx.db.txWaiters); i++ { + txw := tx.db.txWaiters[i] + // in practice this probably never matters, but theoretically the + // goroutine that's waiting on the condition variable may + // not have performed its first test on len(txw.waitingOn) yet. + delete(txw.waitingOn, tx) + // let it know we're done. we've still got db.mu.lock, so it won't + // happen just yet, but it'll be able to continue. + if len(txw.waitingOn) == 0 { + // remove us from the db's list + copy(db.txWaiters[i:], db.txWaiters[i+1:]) + db.txWaiters = db.txWaiters[:len(db.txWaiters)-1] + close(txw.ready) + // decrement i so we don't skip an entry we just copied in to [i] + i-- + } + } // Disassociate from db. tx.db = nil - // Write pages from WAL to DB. - // TODO(bbj): Move this to an async goroutine. - if len(db.txs) == 0 && db.walSize() > db.cfg.MinWALCheckpointSize { - if err := db.checkpoint(); err != nil { - return fmt.Errorf("checkpoint: %w", err) - } + if checkpoint { + // We need to run a checkpoint. This can be semi-asynchronous. + // It needs to wait until every existing transaction has finished, + // because every existing transaction could want to look up pages + // which are in the database before our operations, but which should + // now be in the WAL. We want them to use the WAL instead. + // fmt.Printf("possibly-async checkpoint...\n") + db.afterCurrentTx(func() { + // We still hold db.rwmu here. checkpoint unlocks it when it's + // ready. + // fmt.Printf("checkpoint starting\n") + if err := db.checkpoint(); err != nil { + db.logger.Errorf("async checkpoint: %v", err) + } + }) } - return nil } @@ -547,14 +791,9 @@ func (db *DB) readDBPage(pgno uint32) ([]byte, error) { return db.data[offset : offset+PageSize], nil } -// baseWALID returns the WAL ID stored in the database file meta page. -func (db *DB) baseWALID() int64 { - return readMetaWALID(db.data) -} - // readWALPageByID reads a WAL page by WAL ID. func (db *DB) readWALPageByID(id int64) ([]byte, error) { - return db.readWALPageAt(int(id - db.baseWALID() - 1)) + return db.readWALPageAt(int(id - db.baseWALID - 1)) } // readWALPageAt reads the i-th page in the WAL file. diff --git a/rbf/db_test.go b/rbf/db_test.go index 56170d6eb..c0eb3b3c7 100644 --- a/rbf/db_test.go +++ b/rbf/db_test.go @@ -13,6 +13,7 @@ import ( _ "net/http/pprof" + "github.com/felixge/fgprof" "github.com/molecula/featurebase/v2/rbf" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "golang.org/x/sync/errgroup" @@ -290,7 +291,8 @@ func TestDB_MultiTx(t *testing.T) { time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) - for i := 0; i < rand.Intn(1000); i++ { + n := rand.Intn(500) + 500 + for i := 0; i < n; i++ { v := rand.Intn(1 << 20) if _, err := tx.Contains("x", uint64(v)); err != nil { return err @@ -315,7 +317,8 @@ func TestDB_MultiTx(t *testing.T) { } defer tx.Rollback() - for j := 0; j < rand.Intn(100); j++ { + n := rand.Intn(90) + 10 + for j := 0; j < n; j++ { v := rand.Intn(1 << 20) if _, err := tx.Add("x", uint64(v)); err != nil { t.Fatal(err) @@ -336,6 +339,119 @@ func TestDB_MultiTx(t *testing.T) { } } +// premake pool of random values +const randPool = (1 << 18) + +// benchmarkOneCheckpoint +func benchmarkOneCheckpoint(b *testing.B, randInts []int) { + cfg := rbfcfg.NewDefaultConfig() + // extremely low to force checkpointing + cfg.MinWALCheckpointSize = rbf.PageSize * 16 + cfg.MaxWALCheckpointSize = rbf.PageSize * 64 + var _ rbfcfg.Config + db := MustOpenDB(b, cfg) + defer MustCloseDB(b, db) + + // Run multiple readers in separate goroutines. + ctx, cancel := context.WithCancel(context.Background()) + g, ctx := errgroup.WithContext(ctx) + for i := 0; i < 8; i++ { + i := i + g.Go(func() error { + for { + if ctx.Err() != nil { + return nil // cancelled, return no error + } else if err := func() error { + tx, err := db.Begin(false) + if err != nil { + return err + } + defer tx.Rollback() + + time.Sleep(time.Duration(rand.Intn(int(3 * time.Millisecond)))) + + times := rand.Intn(1000) + 1 + for j := 0; j < times; j++ { + v := randInts[((i<<10)+j)%(randPool-1)] + if _, err := tx.Contains("x", uint64(v)); err != nil { + return err + } + } + return nil + }(); err != nil { + return err + } + // time.Sleep(time.Duration(rand.Intn(int(3 * time.Millisecond)))) + } + }) + } + + // Continuously set/clear bits while readers are executing. + next := 0 + for i := 0; i < 1000; i++ { + func() { + tx, err := db.Begin(true) + if err != nil { + b.Fatal(err) + } + defer tx.Rollback() + + times := rand.Intn(100) + for j := 0; j < times; j++ { + v := randInts[next] + next = (next + 1) % (randPool - 1) + if j&7 == 0 { + // some removes but they're less frequent + if _, err := tx.Remove("x", uint64(v)); err != nil { + b.Fatal(err) + } + } else { + if _, err := tx.Add("x", uint64(v)); err != nil { + b.Fatal(err) + } + } + + } + if err := tx.Commit(); err != nil { + b.Fatal(err) + } + }() + } + + // Stop readers & wait. + cancel() + if err := g.Wait(); err != nil { + b.Fatal(err) + } +} + +func BenchmarkDbCheckpoint(b *testing.B) { + out, err := os.Create("cp.out") + if err != nil { + b.Fatalf("creating log file: %v", err) + } + done := fgprof.Start(out, fgprof.FormatPprof) + b.StopTimer() + // premake these because otherwise it's >5% of CPU in the reads + randInts := make([]int, randPool) + for i := range randInts { + v1, v2 := rand.Intn(1<<24), rand.Intn(1<<24) + // minimum gives us a skewed distribution which makes lower values more + // likely than higher values, so we get a mix of container types + if v1 < v2 { + randInts[i] = v1 + } else { + randInts[i] = v2 + } + } + b.StartTimer() + for i := 0; i < b.N; i++ { + benchmarkOneCheckpoint(b, randInts) + } + b.StopTimer() + done() +} + // better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests. func TestMain(m *testing.M) { l, err := net.Listen("tcp", ":0") diff --git a/rbf/rbf_test.go b/rbf/rbf_test.go index 48f16335a..4071a1a95 100644 --- a/rbf/rbf_test.go +++ b/rbf/rbf_test.go @@ -11,6 +11,7 @@ import ( "sort" "testing" + "github.com/molecula/featurebase/v2/logger" "github.com/molecula/featurebase/v2/rbf" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "github.com/molecula/featurebase/v2/testhook" @@ -65,6 +66,13 @@ func NewDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB { // MustOpenDB returns a db opened on a temporary file. On error, fail test. func MustOpenDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB { tb.Helper() + if len(cfg) == 0 || cfg[0] == nil { + newconf := rbfcfg.NewDefaultConfig() + newconf.Logger = logger.NewLogfLogger(tb) + cfg = []*rbfcfg.Config{newconf} + } else if cfg[0].Logger == nil { + cfg[0].Logger = logger.NewLogfLogger(tb) + } db := NewDB(tb, cfg...) if err := db.Open(); err != nil { tb.Fatal(err) @@ -78,7 +86,14 @@ func MustCloseDB(tb testing.TB, db *rbf.DB) { tb.Helper() if err := db.Check(); err != nil && err != rbf.ErrClosed { tb.Fatal(err) - } else if n := db.TxN(); n != 0 { + } + MustCloseDBNoCheck(tb, db) +} + +// MustCloseDBNoCheck closes db. On error, fail test. +func MustCloseDBNoCheck(tb testing.TB, db *rbf.DB) { + tb.Helper() + if n := db.TxN(); n != 0 { tb.Fatalf("db still has %d active transactions; must closed before closing db", n) } else if err := db.Close(); err != nil && err != rbf.ErrClosed { tb.Fatal(err) diff --git a/rbf/tx.go b/rbf/tx.go index 38fed48f6..5c6c7f7c6 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -109,20 +109,25 @@ func (tx *Tx) Commit() error { // future plan: after checkpoint is moved to background // or not every removeTx, then we can move the // tx.db.rootRecords = tx.rootRecords into removeTx(). - + // + // ... or maybe not: let's do that part here, and then removeTx + // may or may not start a checkpoint, possibly asynchronously. + // // avoid race detector firing on a write race here - // vs the read of rootRecords at db.Begin() + // vs the read of rootRecords at db.Begin(), then release + // the lock, because we need removeTx to grab the lock to + // work, but if it wants to checkpoint, it wants to be able to return + // to us here and still be holding the lock. tx.db.mu.Lock() - defer tx.db.mu.Unlock() tx.db.rootRecords = tx.rootRecords tx.db.pageMap = tx.pageMap tx.db.walPageN = tx.walPageN - return tx.db.removeTx(tx) + tx.db.mu.Unlock() } - // Disconnect transaction from DB. tx.db.mu.Lock() defer tx.db.mu.Unlock() + // Disconnect transaction from DB. return tx.db.removeTx(tx) } @@ -736,6 +741,27 @@ func (tx *Tx) Check() error { return nil } +func (tx *Tx) checkPage(pgno, parent, typ uint32) error { + switch typ { + case PageTypeBranch: + return tx.checkBranchPage(pgno, parent, typ) + default: + return nil + } +} + +func (tx *Tx) checkBranchPage(pgno, parent, typ uint32) error { + page, _, err := tx.readPage(pgno) + if err != nil { + return err + } + + if readCellN(page) == 0 { + return fmt.Errorf("branch page %d is empty", pgno) + } + return nil +} + // checkPageAllocations ensures that all pages are either in-use or on the freelist. func (tx *Tx) checkPageAllocations() error { freePageSet, err := tx.freePageSet() @@ -825,7 +851,7 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) { // Traverse freelist and mark pages as in-use. if err := tx.walkTree(readMetaFreelistPageNo(tx.meta[:]), 0, func(pgno, parent, typ uint32) error { m[pgno] = struct{}{} - return nil + return tx.checkPage(pgno, parent, typ) }); err != nil { return m, err } @@ -841,7 +867,8 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) { if err := tx.walkTree(pgno.(uint32), 0, func(pgno, parent, typ uint32) error { m[pgno] = struct{}{} - return nil + + return tx.checkPage(pgno, parent, typ) }); err != nil { return m, err } diff --git a/rbf/tx_test.go b/rbf/tx_test.go index 1004437a3..5be3a50ae 100644 --- a/rbf/tx_test.go +++ b/rbf/tx_test.go @@ -2,8 +2,11 @@ package rbf_test import ( + "encoding/binary" "fmt" "math/rand" + "os" + "strings" "sync" "testing" "time" @@ -433,6 +436,52 @@ func TestTx_DeallocateToFreeList(t *testing.T) { } } +func TestTx_Remove(t *testing.T) { + t.Parallel() + + db := MustOpenDB(t) + defer MustCloseDB(t, db) + + tx := MustBegin(t, db, true) + defer tx.Rollback() + + if err := tx.CreateBitmap("x"); err != nil { + t.Fatal(err) + } + + // Insert large array values. + var values []uint64 + for i := 0; i < 1000; i++ { + for j := 0; j < rbf.ArrayMaxSize; j++ { + v := uint64((i << 16) + j) + values = append(values, v) + + if _, err := tx.Add("x", v); err != nil { + t.Fatalf("Add(%d) err=%q", v, err) + } + } + } + + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + + tx = MustBegin(t, db, true) + defer tx.Rollback() + + // Remove all array values. + for _, i := range rand.Perm(len(values)) { + v := values[i] + if _, err := tx.Remove("x", v); err != nil { + t.Fatalf("Remove(%d) err=%q", v, err) + } + } + + if err := tx.Commit(); err != nil { + t.Fatal(err) + } +} + func TestTx_AddRemove_Quick(t *testing.T) { if testing.Short() { t.Skip("-short enabled, skipping") @@ -770,3 +819,83 @@ func TestTx_DeleteBitmapsWithPrefix(t *testing.T) { checkInfos() } + +func TestTx_Check(t *testing.T) { + t.Run("EmptyBranchPage", func(t *testing.T) { + t.Parallel() + + db := MustOpenDB(t) + defer MustCloseDBNoCheck(t, db) + tx := MustBegin(t, db, true) + defer tx.Rollback() + + if err := tx.CreateBitmap("x"); err != nil { + t.Fatal(err) + } + + // Insert enough array containers to split page. + for i := 0; i < 1000; i++ { + if _, err := tx.Add("x", uint64(i<<16)); err != nil { + t.Fatalf("Add(%d) err=%q", i<<16, err) + } + } + + // Read page types for all pages. + infos, err := tx.PageInfos() + if err != nil { + t.Fatal(err) + } + + // Commit & checkpoint to flush to the data file. + if err := tx.Commit(); err != nil { + t.Fatal(err) + } else if err := db.Checkpoint(); err != nil { + t.Fatal(err) + } + + // Corrupt first branch page found by zeroing out the cell count. + var pgno uint32 + for _, info := range infos { + if info, ok := info.(*rbf.BranchPageInfo); ok { + pgno = info.Pgno + page := mustReadPage(t, db.DataPath(), pgno) + binary.BigEndian.PutUint16(page[8:10], 0) // zero cell count + mustWritePage(t, db.DataPath(), pgno, page) + break + } + } + + // Verify that check now returns an error. + if err := db.Check(); err == nil || !strings.Contains(err.Error(), fmt.Sprintf("branch page %d is empty", pgno)) { + t.Fatalf("unexpected error: %#v", err) + } + }) +} + +func mustReadPage(tb testing.TB, path string, pgno uint32) []byte { + tb.Helper() + f, err := os.Open(path) + if err != nil { + tb.Fatal(err) + } + defer f.Close() + + buf := make([]byte, rbf.PageSize) + if _, err := f.ReadAt(buf, int64(pgno)*rbf.PageSize); err != nil { + tb.Fatal(err) + } + return buf +} + +func mustWritePage(tb testing.TB, path string, pgno uint32, buf []byte) { + tb.Helper() + f, err := os.OpenFile(path, os.O_WRONLY, 0666) + if err != nil { + tb.Fatal(err) + } + defer f.Close() + + if _, err := f.WriteAt(buf, int64(pgno)*rbf.PageSize); err != nil { + tb.Fatal(err) + } +} diff --git a/server.go b/server.go index ba5891863..08b3b0fa5 100644 --- a/server.go +++ b/server.go @@ -484,7 +484,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { } s.holder = NewHolder(path, s.holderConfig) s.holder.Stats.SetLogger(s.logger) - s.holder.Logger.Infof("RowCacheOn: %v", s.holderConfig.RowcacheOn) cwd, err := os.Getwd() if err != nil { return nil, err diff --git a/server/config.go b/server/config.go index 7801b8ce0..df9dafdee 100644 --- a/server/config.go +++ b/server/config.go @@ -4,15 +4,18 @@ package server import ( "context" "fmt" + "io" "log" "net" "net/url" + "os" + "path/filepath" "runtime" "strconv" "strings" "time" - "github.com/molecula/featurebase/v2/auth" + "github.com/molecula/featurebase/v2/authz" petcd "github.com/molecula/featurebase/v2/etcd" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "github.com/molecula/featurebase/v2/storage" @@ -203,9 +206,9 @@ type Config struct { // "rbf". Storage *storage.Config `toml:"storage"` - // RowcacheOn, if true, turns on the row cache for all storage backends. - // The default is now off because it makes rbf queries faster and uses - // much less memory. + // RowcacheOn permanently disabled. No longer useful w/ RBF. Left + // for backward compatibility but will be removed in a future + // version. RowcacheOn bool `toml:"rowcache-on"` // RBFConfig defines all externally configurable RBF flags. @@ -233,8 +236,23 @@ type Config struct { // Toggles /schema/details endpoint. If off, it returns empty. SchemaDetailsOn bool `toml:"schema-details-on"` - // Enable AuthZ/AuthN - Auth auth.Auth `toml:"auth"` + Auth Auth +} + +type Auth struct { + // Enable AuthZ/AuthN for featurebase server + Enable bool `toml:"enable"` + + ClientId string `toml:"client-id"` + ClientSecret string `toml:"client-secret"` + AuthorizeURL string `toml:"authorize-url"` + TokenURL string `toml:"token-url"` + GroupEndpointURL string `toml:"group-endpoint-url"` + LogoutURL string `toml:"logout-url"` + Scopes []string `toml:"scopes"` + HashKey string `toml:"hash-key"` + BlockKey string `toml:"block-key"` + PermissionsFile string `toml:"permissions"` } // Namespace returns the namespace to use based on the Future flag. @@ -599,9 +617,9 @@ func lookupAddr(ctx context.Context, resolver *net.Resolver, host string) (strin return addrs[0].String(), nil } -func (c *Config) ValidateAuth() ([]error, error) { +func (c *Config) ValidateAuth() (errors []error) { if !c.Auth.Enable { - return []error{}, nil + return } authConfig := map[string]string{ "ClientId": c.Auth.ClientId, @@ -609,16 +627,23 @@ func (c *Config) ValidateAuth() ([]error, error) { "AuthorizeURL": c.Auth.AuthorizeURL, "TokenURL": c.Auth.TokenURL, "GroupEndpointURL": c.Auth.GroupEndpointURL, - "ScopeURL": c.Auth.ScopeURL, + "LogoutURL": c.Auth.LogoutURL, + "HashKey": c.Auth.HashKey, + "BlockKey": c.Auth.BlockKey, } - errors := make([]error, 0) for name, value := range authConfig { if value == "" { errors = append(errors, fmt.Errorf("empty string for auth config %s", name)) continue } + if name == "HashKey" || name == "BlockKey" { + if len(value) != 64 { + errors = append(errors, fmt.Errorf("invalid key length for %s. exp %d, got %d", name, 64, len(value))) + } + } + if strings.Contains(name, "URL") { _, err := url.ParseRequestURI(value) if err != nil { @@ -627,17 +652,101 @@ func (c *Config) ValidateAuth() ([]error, error) { } } } - if len(errors) > 0 { - return errors, fmt.Errorf("there were errors validating config") + + if len(c.Auth.Scopes) == 0 { + errors = append(errors, fmt.Errorf("must provide scope for authentication with IdP - for access and refresh token")) } - return errors, nil + + return errors +} + +func (c *Config) ValidatePermissions(permsFile io.Reader) (errors []error) { + + var p authz.GroupPermissions + if err := p.ReadPermissionsFile(permsFile); err != nil { + return append(errors, err) + } + + if len(p.Permissions) == 0 { + return append(errors, fmt.Errorf("no group permissions found in permissions file: %s", c.Auth.PermissionsFile)) + } + + for groupId, indexPerm := range p.Permissions { + if groupId == "" { + errors = append(errors, fmt.Errorf("empty string for group id in permissions file %s", c.Auth.PermissionsFile)) + continue + } + + for index, perm := range indexPerm { + if index == "" { + errors = append(errors, fmt.Errorf("empty string for index for group id %s in permissions file %s ", groupId, c.Auth.PermissionsFile)) + continue + } + + if perm == "" { + errors = append(errors, fmt.Errorf("empty string for permission for group id %s and index %s in permissions file %s", groupId, index, c.Auth.PermissionsFile)) + continue + } + + if !((perm == "write") || (perm == "read")) { + errors = append(errors, fmt.Errorf("not a valid permission %s for group id %s and index %s in permissions file %s; expected permissions are read or write", perm, groupId, index, c.Auth.PermissionsFile)) + continue + } + } + } + + if p.Admin == "" { + errors = append(errors, fmt.Errorf("empty string for admin in permissions file: %s", c.Auth.PermissionsFile)) + + } + + return errors +} + +func (c *Config) ValidatePermissionsFile() (err error) { + + if c.Auth.PermissionsFile == "" { + return fmt.Errorf("empty string for auth config permissions file") + } + + fileExt := filepath.Ext(c.Auth.PermissionsFile) + if (fileExt != ".yaml") && (fileExt != ".yml") { + return fmt.Errorf("invalid file extension for auth config permissions file: %s", c.Auth.PermissionsFile) + } + return } func (c *Config) MustValidateAuth() { - if errors, err := c.ValidateAuth(); err != nil { - for _, e := range errors { + + errorsAuth := c.ValidateAuth() + if len(errorsAuth) > 0 { + for _, e := range errorsAuth { log.Println(e) } - log.Fatal(err) + } + + var errorsPerm []error + errorsPermFile := c.ValidatePermissionsFile() + if errorsPermFile == nil { + permsFile, err := os.Open(c.Auth.PermissionsFile) + if err != nil { + log.Println(err) + } + + defer permsFile.Close() + + errorsPerm = c.ValidatePermissions(permsFile) + if len(errorsPerm) > 0 { + for _, e := range errorsPerm { + log.Println(e) + } + } + + } else { + log.Println(errorsPermFile) + } + + if len(errorsAuth) > 0 || len(errorsPerm) > 0 || errorsPermFile != nil { + log.Fatal(fmt.Errorf("there were errors validating authN/authZ config and/or permissions")) } } diff --git a/server/config_internal_test.go b/server/config_internal_test.go index 7c762b23e..4af9007fe 100644 --- a/server/config_internal_test.go +++ b/server/config_internal_test.go @@ -8,8 +8,6 @@ import ( "os" "strings" "testing" - - "github.com/molecula/featurebase/v2/auth" ) type addrs struct{ bind, advertise string } @@ -281,19 +279,25 @@ func TestConfig_validateAddrsGRPC(t *testing.T) { func TestConfig_validateAuth(t *testing.T) { errorMesgEmpty := "empty string" errorMesgURL := "invalid URL" + errorMesgScope := "must provide scope" + errorMesgKey := "invalid key length" validTestURL := "https://url.com/" validClientID := "clientid" validClientSecret := "clientSecret" - notValidURL := "not-a-url" + validKey := "3db6665be8b860af422155acf2346d4fcb46678fca42e60d934abe0b7ce43600" + invalidURL := "not-a-url" emptyString := "" + validStringSlice := []string{"https://graph.microsoft.com/.default", "offline_access"} + validString := "asdfqwer1234asdfzxcv" + var emptySlice []string + enable := true disable := false tests := []struct { expErrs []string - input auth.Auth + input Auth }{ - { // Auth enabled, all configs are set to empty string []string{ @@ -303,161 +307,109 @@ func TestConfig_validateAuth(t *testing.T) { errorMesgEmpty, errorMesgEmpty, errorMesgEmpty, + errorMesgEmpty, + errorMesgEmpty, }, - auth.Auth{ + Auth{ Enable: enable, ClientId: emptyString, ClientSecret: emptyString, AuthorizeURL: emptyString, TokenURL: emptyString, GroupEndpointURL: emptyString, - ScopeURL: emptyString, + LogoutURL: emptyString, + Scopes: validStringSlice, + HashKey: emptyString, + BlockKey: emptyString, }, }, { - // Auth enabled, some configs are set to empty string + // Auth enabled, keys are invalid length []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, + errorMesgKey, + errorMesgKey, }, - auth.Auth{ - Enable: enable, - ClientId: validClientID, - ClientSecret: emptyString, - AuthorizeURL: emptyString, - TokenURL: emptyString, - GroupEndpointURL: emptyString, - ScopeURL: emptyString, - }, - }, - { - // Auth enabled, some configs are set to empty string - []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - }, - auth.Auth{ - Enable: enable, - ClientId: emptyString, - ClientSecret: validClientSecret, - AuthorizeURL: emptyString, - TokenURL: emptyString, - GroupEndpointURL: emptyString, - ScopeURL: emptyString, - }, - }, - { - // Auth enabled, some configs are set to empty string - []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - }, - auth.Auth{ - Enable: enable, - ClientId: validClientID, - ClientSecret: validClientSecret, - AuthorizeURL: emptyString, - TokenURL: emptyString, - GroupEndpointURL: emptyString, - ScopeURL: emptyString, - }, - }, - { - // Auth enabled, some configs are set to empty string - []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - }, - auth.Auth{ + Auth{ Enable: enable, ClientId: validClientID, ClientSecret: validClientSecret, AuthorizeURL: validTestURL, - TokenURL: emptyString, - GroupEndpointURL: emptyString, - ScopeURL: emptyString, - }, - }, - { - // Auth enabled, some configs are set to empty string - []string{ - errorMesgEmpty, - errorMesgEmpty, - }, - auth.Auth{ - Enable: enable, - ClientId: validClientID, - ClientSecret: validClientSecret, - AuthorizeURL: validTestURL, - TokenURL: validTestURL, - GroupEndpointURL: emptyString, - ScopeURL: emptyString, - }, - }, - { - // Auth enabled, some strings are set to invalid URL - []string{ - errorMesgURL, - }, - auth.Auth{ - Enable: enable, - ClientId: validClientID, - ClientSecret: validClientSecret, - AuthorizeURL: notValidURL, TokenURL: validTestURL, GroupEndpointURL: validTestURL, - ScopeURL: validTestURL, + LogoutURL: validTestURL, + Scopes: validStringSlice, + HashKey: validString, + BlockKey: validString, }, }, { - // Auth enabled, some strings are set to invalid URL + // Auth enabled, some URLs are set to invalid URL []string{ errorMesgURL, errorMesgURL, + errorMesgURL, }, - auth.Auth{ + Auth{ Enable: enable, ClientId: validClientID, ClientSecret: validClientSecret, AuthorizeURL: validTestURL, - TokenURL: notValidURL, - GroupEndpointURL: notValidURL, - ScopeURL: validTestURL, + TokenURL: invalidURL, + GroupEndpointURL: invalidURL, + LogoutURL: invalidURL, + Scopes: validStringSlice, + HashKey: validKey, + BlockKey: validKey, + }, + }, + { + // Auth enabled, all configs are set properly except scope + []string{ + errorMesgScope, + }, + Auth{ + Enable: enable, + ClientId: validClientID, + ClientSecret: validClientSecret, + AuthorizeURL: validTestURL, + TokenURL: validTestURL, + GroupEndpointURL: validTestURL, + LogoutURL: validTestURL, + Scopes: emptySlice, + HashKey: validKey, + BlockKey: validKey, }, }, { // Auth enabled, all configs are set properly []string{}, - auth.Auth{ + Auth{ Enable: enable, ClientId: validClientID, ClientSecret: validClientSecret, AuthorizeURL: validTestURL, TokenURL: validTestURL, GroupEndpointURL: validTestURL, - ScopeURL: validTestURL, + LogoutURL: validTestURL, + Scopes: validStringSlice, + HashKey: validKey, + BlockKey: validKey, }, }, { - // Auth disabled, all configs are set to empty string + // Auth disabled, some configs are set to values []string{}, - auth.Auth{ + Auth{ Enable: disable, ClientId: emptyString, - ClientSecret: emptyString, + ClientSecret: validString, AuthorizeURL: emptyString, TokenURL: emptyString, - GroupEndpointURL: emptyString, - ScopeURL: emptyString, + GroupEndpointURL: invalidURL, + LogoutURL: validTestURL, + Scopes: validStringSlice, + HashKey: validKey, + BlockKey: emptyString, }, }, } @@ -467,9 +419,9 @@ func TestConfig_validateAuth(t *testing.T) { c := NewConfig() c.Auth = test.input - errors, err := c.ValidateAuth() + errors := c.ValidateAuth() if len(test.expErrs) > 0 { - if err == nil { + if errors == nil { t.Fatal("expected errors, but none were found") } } @@ -487,3 +439,113 @@ func TestConfig_validateAuth(t *testing.T) { }) } } + +func TestConfig_validatePermissions(t *testing.T) { + permissions0 := `` + + permissions1 := `user-groups: + "": + "test": "read" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions2 := `user-groups: + "dca35310-ecda-4f23-86cd-876aee559900": + "": "write" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions3 := `user-groups: + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions4 := `user-groups: + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "readwrite" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions5 := `user-groups: + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "read"` + + tests := []struct { + err string + input string + }{ + { + "no group permissions found in permissions file", + permissions0, + }, + { + "empty string for group id", + permissions1, + }, + { + "empty string for index", + permissions2, + }, + { + "empty string for permission", + permissions3, + }, + { + "not a valid permission", + permissions4, + }, + { + "empty string for admin in permissions file", + permissions5, + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + + c := NewConfig() + c.Auth.PermissionsFile = "test.yaml" + + permFile := strings.NewReader(test.input) + errors := c.ValidatePermissions(permFile) + + if errors == nil { + t.Fatal("expected errors, but none were found") + } + + for _, err := range errors { + if !strings.Contains(err.Error(), test.err) { + t.Errorf("expected error to contain %s, but got %s", test.err, err.Error()) + + } + } + }) + } +} + +func TestConfig_validatePermissionsFilename(t *testing.T) { + + tests := []struct { + err string + input string + }{ + { + "empty string for auth config permissions file", + "", + }, + { + "invalid file extension for auth config permissions file", + "permissions.txt", + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + c := NewConfig() + c.Auth.PermissionsFile = test.input + + if err := c.ValidatePermissionsFile(); err != nil { + if !strings.Contains(err.Error(), test.err) { + t.Errorf("expected error to contain %s, but got %s", test.err, err.Error()) + } + } + }) + } +} diff --git a/server/server.go b/server/server.go index d06b52ba8..387a34deb 100644 --- a/server/server.go +++ b/server/server.go @@ -29,6 +29,8 @@ import ( "golang.org/x/sync/errgroup" pilosa "github.com/molecula/featurebase/v2" + "github.com/molecula/featurebase/v2/authn" + "github.com/molecula/featurebase/v2/authz" "github.com/molecula/featurebase/v2/boltdb" "github.com/molecula/featurebase/v2/encoding/proto" petcd "github.com/molecula/featurebase/v2/etcd" @@ -83,6 +85,8 @@ type Command struct { pgserver *PostgresServer serverOptions []pilosa.ServerOption + + auth *authn.Auth } type CommandOption func(c *Command) error @@ -224,10 +228,6 @@ func (m *Command) Start() (err error) { return errors.Wrap(err, "setting resource limits") } - if m.Config.Auth.Enable { - m.Config.MustValidateAuth() - } - // Initialize server. if err = m.Server.Open(); err != nil { return errors.Wrap(err, "opening server") @@ -489,7 +489,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerClusterName(m.Config.Cluster.Name), pilosa.OptServerSerializer(proto.Serializer{}), pilosa.OptServerStorageConfig(m.Config.Storage), - pilosa.OptServerRowcacheOn(m.Config.RowcacheOn), + pilosa.OptServerRowcacheOn(false), pilosa.OptServerRBFConfig(m.Config.RBFConfig), pilosa.OptServerMaxQueryMemory(m.Config.MaxQueryMemory), pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength), @@ -530,6 +530,26 @@ func (m *Command) SetupServer() error { return errors.Wrap(err, "new grpc server") } + if m.Config.Auth.Enable { + m.Config.MustValidateAuth() + permsFile, err := os.Open(m.Config.Auth.PermissionsFile) + if err != nil { + return err + } + defer permsFile.Close() + + var p authz.GroupPermissions + if err = p.ReadPermissionsFile(permsFile); err != nil { + return err + } + + ac := m.Config.Auth + m.auth, err = authn.NewAuth(m.logger, m.listenURI.String(), ac.Scopes, ac.AuthorizeURL, ac.TokenURL, ac.GroupEndpointURL, ac.LogoutURL, ac.ClientId, ac.ClientSecret, ac.HashKey, ac.BlockKey) + if err != nil { + return errors.Wrap(err, "instantiating authN object") + } + } + m.Handler, err = http.NewHandler( http.OptHandlerAllowedOrigins(m.Config.Handler.AllowedOrigins), http.OptHandlerAPI(m.API), @@ -539,6 +559,7 @@ func (m *Command) SetupServer() error { http.OptHandlerListener(m.ln, m.Config.Advertise), http.OptHandlerCloseTimeout(m.closeTimeout), http.OptHandlerMiddleware(m.grpcServer.middleware(m.Config.Handler.AllowedOrigins)), + http.OptHandlerAuth(m.auth), ) return errors.Wrap(err, "new handler") } diff --git a/txfactory.go b/txfactory.go index 62f527653..12fa12981 100644 --- a/txfactory.go +++ b/txfactory.go @@ -242,10 +242,15 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) { } // qcx.write reflects the top executor determination - // if a write will be done at the end, so we upgrade - // the "local" read Tx to be writes, so that they - // don't deadlock against themselves. - o.Write = o.Write || qcx.write + // if a write will be happen at some point, in which case, to avoid + // locking problems with multi-shard things, we (probably incorrectly) + // treat every Tx as its own individual separate Tx. + // + // But we still want to open non-write transactions individually, we + // just can't recycle them (because write operations will come in and + // we want them to work and commit right away so we're not holding a write + // lock for long). + writeLogic := o.Write || qcx.write // In general, we make ALL write transactions local, and never reuse them // below. Previously this was to help lmdb. @@ -273,7 +278,7 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) { return *qcx.RequiredForAtomicWriteTx, NoopFinisher, nil } - if !o.Write && qcx.Grp != nil { + if !writeLogic && qcx.Grp != nil { // read, with a group in place. finisher = func(perr *error) {} // finisher is a returned value diff --git a/view.go b/view.go index f3bd27b3a..5a8e23ae1 100644 --- a/view.go +++ b/view.go @@ -619,7 +619,9 @@ func (v *view) bitDepth(shards []uint64) (uint64, error) { var maxBitDepth uint64 for _, shard := range shards { + v.mu.RLock() frag, ok := v.fragments[shard] + v.mu.RUnlock() if !ok || frag == nil { continue }