scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-10 15:01:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch '54mir/protect-endpoints' into queryLoggerSetup

Browse source
This commit is contained in:
Samir Patel 2021-12-22 15:12:31 -06:00
commit 684c408b93
47 changed files with 2930 additions and 378 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
api.go
View file

@ -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

25
auth/auth.go
View file

@ -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"`
}

296
authn/authenticate.go Normal file
View file

@ -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,
}
}

118
authn/authenticate_internal_test.go Normal file
View file

@ -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)
}
})
}

120
authz/authorization.go Normal file
View file

@ -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
}

318
authz/authorization_test.go Normal file
View file

@ -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)
}
})
}
}

28
cmd/keygen.go Normal file
View file

@ -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
}

4
cmd/root.go
View file

@ -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

31
ctl/keygen.go Normal file
View file

@ -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
}

9
ctl/rbf_pages.go
View file

@ -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)
}
}

8
ctl/server.go
View file

@ -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.")
}

89
executor.go
View file

@ -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.

7
field.go
View file

@ -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
}

77
fragment.go
View file

@ -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.

5
go.mod
View file

@ -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

6
go.sum
View file

@ -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=

88
http/handler.go
View file

@ -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)
}

410
http/handler_internal_test.go
View file

@ -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)
})
}
}
}

9
install/featurebase.conf
View file

@ -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 = ""
# logout-url = ""
# scopes = ["", ""]
# hash-key = ""
# block-key = ""
# permissions = ""

75
lattice/src/App.tsx
View file

@ -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<string>(
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 (
<MuiThemeProvider theme={theme === 'light' ? lightTheme : darkTheme}>
<CssBaseline />
<Header onToggleTheme={onToggleTheme} />
<div className={css.container}>
<div className={css.layout}>
<Nav />
<div className={css.mainContent}>
<BrowserRouter>
{auth.isLoading ? (
// Loading, retreiving auth status
<div></div>
) : (
// Loading done, display app based on auth status
<MuiThemeProvider theme={lightTheme}>
{auth.isAuthOn ? (
// Auth is on, hide the routes with PrivateRoute
<Switch>
<Route exact path="/" component={Home} />
<Route path="/tables/:id?" component={MoleculaTablesContainer} />
<Route exact path="/query" component={QueryContainer} />
<Route exact path="/querybuilder" component={QueryBuilderContainer} />
<Route component={NotFound} />
<Route exact path="/signin" render={(props) => <Signin {...props}></Signin>} />
<PrivateRoute path="/" component={Main} />
</Switch>
</div>
</div>
</div>
</MuiThemeProvider>
) : (
// Auth is off, all routes are accessible
<Route path="/" component={Main} />
)}
</MuiThemeProvider>
)}
</BrowserRouter>
);
}
};
export default App;

56
lattice/src/App/AuthFlow/AuthFlow.module.scss Normal file
View file

@ -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;
}

20
lattice/src/App/AuthFlow/SignInButton.tsx Normal file
View file

@ -0,0 +1,20 @@
import React from 'react';
import { Button } from '@material-ui/core';
interface Props {
children?: React.ReactNode;
}
const SignInButton: React.FC<Props> = ({ children }) => {
const signinOnClick = (e) => {
window.location.href = '/login';
};
return (
<Button variant="contained" color="primary" size="large" onClick={signinOnClick} fullWidth>
Sign in
</Button>
);
};
export default SignInButton;

20
lattice/src/App/AuthFlow/SignOutButton.tsx Normal file
View file

@ -0,0 +1,20 @@
import React from 'react';
import { Button } from '@material-ui/core';
interface Props {
children?: React.ReactNode;
}
const SignOutButton: React.FC<Props> = ({ children }) => {
const signoutOnClick = (e) => {
window.location.href = '/logout';
};
return (
<Button variant="contained" color="secondary" onClick={signoutOnClick}>
Sign out
</Button>
);
};
export default SignOutButton;

30
lattice/src/App/AuthFlow/Signin.tsx Normal file
View file

@ -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 = () => (
<Card>
<CardHeader subheader={'Sign in to continue'} />
<CardContent>
<SignInButton />
</CardContent>
</Card>
);
return (
<div className={css.main}>
<div className={css.loginForm}>
<div className={css.logoContainer}>
<MLogo className={css.logo} />
</div>
{renderLoginForm()}
</div>
</div>
);
}
export default Signin;

1
lattice/src/App/AuthFlow/index.ts Normal file
View file

@ -0,0 +1 @@
export * from './Signin';

58
lattice/src/Main.tsx Normal file
View file

@ -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<string>(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 (
<div>
<MuiThemeProvider theme={theme === 'light' ? lightTheme : darkTheme}>
<CssBaseline />
<Header onToggleTheme={onToggleTheme} />
<div className={css.container}>
<div className={css.layout}>
<Nav />
<div className={css.mainContent}>
<Switch>
<Route exact path="/" component={Home} />
<Route path="/tables/:id?" component={MoleculaTablesContainer} />
<Route exact path="/query" component={QueryContainer} />
<Route exact path="/querybuilder" component={QueryBuilderContainer} />
<Route component={NotFound} />
</Switch>
</div>
</div>
</div>
</MuiThemeProvider>
</div>
);
};
export default Main;

BIN
lattice/src/assets/bg-pattern.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 66 KiB

16
lattice/src/assets/m-bug-alt.svg Normal file
View file

@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<svg viewBox="0 0 276 313" version="1.1" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
<defs>
<polygon id="path-1" points="0 0.35 275.231 0.35 275.231 312.94 0 312.94"></polygon>
</defs>
<g id="Group-3">
<mask id="mask-2" fill="white">
<use xlink:href="#path-1"></use>
</mask>
<g id="Clip-2"></g>
<path d="M275.231,86.005 C275.231,81.175 271.809,75.246 267.625,72.831 L145.222,2.162 C141.038,-0.254 134.193,-0.254 130.01,2.162 L7.606,72.831 C3.423,75.246 0,81.175 0,86.005 L0,227.345 C0,232.175 3.423,238.103 7.606,240.518 L130.01,311.188 C134.193,313.603 141.038,313.603 145.222,311.188 L267.625,240.518 C271.809,238.103 275.231,232.175 275.231,227.345 L275.231,86.005" id="Fill-1" fill="#084E83" mask="url(#mask-2)"></path>
</g>
<path d="M143.926,42.423 C138.489,39.277 131.526,41.13 128.376,46.569 C125.226,52.008 127.082,58.97 132.52,62.119 L216.669,110.849 L216.669,208.127 C216.669,214.412 221.764,219.507 228.049,219.507 C234.334,219.507 239.429,214.412 239.429,208.127 L239.429,97.728 L143.926,42.423" id="Fill-4" fill="#C3D4DD"></path>
<path d="M83.364,72.494 C80.213,77.933 82.07,84.895 87.508,88.044 L171.657,136.774 L171.657,234.053 C171.657,240.338 176.751,245.433 183.036,245.433 C189.321,245.433 194.416,240.338 194.416,234.053 L194.416,123.653 L98.913,68.349 C93.477,65.202 86.513,67.055 83.364,72.494" id="Fill-5" fill="#48B5CD"></path>
<path d="M38.185,98.634 C37.203,100.33 36.708,102.173 36.656,104.003 C36.633,105.008 36.819,208.157 36.819,208.157 C36.819,214.442 41.913,219.537 48.199,219.537 C54.484,219.537 59.578,214.442 59.578,208.157 L59.578,124.13 L81.632,136.915 L81.632,234.127 C81.632,240.412 86.727,245.507 93.012,245.507 C99.297,245.507 104.392,240.412 104.392,234.127 L104.392,150.11 L126.645,163.011 L126.645,260.017 C126.645,266.301 131.739,271.396 138.024,271.396 C144.309,271.396 149.404,266.301 149.404,260.017 L149.404,149.846 L53.735,94.489 C48.299,91.342 41.334,93.196 38.185,98.634" id="Fill-6" fill="#FFFFFF"></path>
</svg>
Side by side

After

Width:  |  Height:  |  Size: 2.1 KiB

13
lattice/src/index.tsx
View file

@ -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(
<Router>
<Route path="/" component={App} />
</Router>,
<React.StrictMode>
<ProvideAuth>
<App />
</ProvideAuth>
</React.StrictMode>,
document.getElementById('root')
);

17
lattice/src/services/eventServices.tsx
View file

@ -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);
}
}
},
},
};

81
lattice/src/services/useAuth.tsx Normal file
View file

@ -0,0 +1,81 @@
import React, { createContext, useContext, useEffect, useState } from 'react';
import { pilosa } from './eventServices';
const authContext = createContext<any>({});
// 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 <authContext.Provider value={auth}>{children}</authContext.Provider>;
}
// 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<IUser | undefined>(undefined);
const [isAuthenticated, setIsAuthenticated] = useState<boolean>(false);
const [isLoading, setIsLoading] = useState<boolean>(true);
const [isAuthOn, setIsAuthOn] = useState<boolean>(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,
};
}

42
lattice/src/shared/Header/Header.tsx
View file

@ -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<HeaderProps> = ({ onToggleTheme }) => {
const theme = useTheme();
const isDark = theme.palette.type === 'dark';
const isDark = theme.palette.type === "dark";
const auth = useAuth();
return (
<AppBar
@ -38,6 +44,22 @@ export const Header: FC<HeaderProps> = ({ onToggleTheme }) => {
/>
</div>
</div>
{auth.isAuthenticated ? (
<div>
{auth.user && (
<Button
style={{
backgroundColor: "transparent",
pointerEvents: "none",
}}
>
{auth.user.username}
</Button>
)}
<SignOutButton />
</div>
) : null}
</div>
</Toolbar>
</AppBar>

33
lattice/src/shared/PrivateRoute/PrivateRoute.tsx Normal file
View file

@ -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 (
<Route
{...rest}
render={(props) => {
if (auth.isAuthenticated) {
// If the user is authenticated, render the component
return <Component {...rest} {...props} />;
} else {
// If the user is not authenticated, redirect to sign in page
return (
<Redirect
to={{
pathname: "/signin",
state: {
from: props.location,
},
}}
/>
);
}
}}
/>
);
}
export default PrivateRoute;

6
rbf/cfg/cfg.go
View file

@ -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 {

22
rbf/cursor.go
View file

@ -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
}

4
rbf/cursor_test.go
View file

@ -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) {

375
rbf/db.go
View file

@ -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.

120
rbf/db_test.go
View file

@ -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")

17
rbf/rbf_test.go
View file

@ -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)

41
rbf/tx.go
View file

@ -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
}

129
rbf/tx_test.go
View file

@ -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)
}
}

1
server.go
View file

@ -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

141
server/config.go
View file

@ -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"))
}
}

290
server/config_internal_test.go
View file

@ -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())
}
}
})
}
}

31
server/server.go
View file

@ -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")
}

15
txfactory.go
View file

@ -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

2
view.go
View file

@ -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
}