diff --git a/.gitignore b/.gitignore index 6758e781b..675ffe9df 100644 --- a/.gitignore +++ b/.gitignore @@ -20,3 +20,4 @@ launch.json __pycache__/ report.xml outputs.json +builds/ \ No newline at end of file diff --git a/.gitlab/.gitlab-ci.yml b/.gitlab/.gitlab-ci.yml index 91e8a5480..6dbd26dc8 100644 --- a/.gitlab/.gitlab-ci.yml +++ b/.gitlab/.gitlab-ci.yml @@ -10,8 +10,9 @@ include: - key: $CI_COMMIT_REF_SLUG paths: - .go/pkg/mod/ + variables: - GOVERSION: "1.16.10" + GOVERSION: "1.16.13" stages: - lint @@ -75,15 +76,34 @@ run go tests: - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"' script: - echo "Running featurebase unit tests..." - - PKG_LIST=$(go list ./... | grep -v internal/clustertests | paste -s -d, -) - - go test -coverprofile=coverage.out -covermode=atomic -coverpkg=${PKG_LIST} ./... - artifacts: - paths: - - coverage.out + - go test ./... +run go tests race: + stage: test + image: golang:$GOVERSION + extends: .go-cache + rules: + - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"' + script: + - echo "Running featurebase race tests..." + - go test -race -timeout=30m ./... + +run go tests shardwidth22: + stage: test + image: golang:$GOVERSION + extends: .go-cache + rules: + - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"' + script: + - echo "Running featurebase race tests..." + - go test -tags=shardwidth22 ./... + +# we do coverage reporting from the future tests because the json +# output is very difficult to human-read. The alternative would be to +# run the regular tests twice and also run the future tests. run go tests future: stage: test - image: golang:1.17.3 + image: golang:1.17.6 extends: .go-cache rules: - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"' @@ -106,7 +126,7 @@ upload to sonarcloud: script: - sonar-scanner -Dsonar.projectKey=molecula_featurebase -Dsonar.organization=molecula -Dsonar.sources=. -Dsonar.host.url=https://sonarcloud.io -Dsonar.go.coverage.reportPaths=coverage.out -Dsonar.go.tests.reportPaths=test-report.out -Dsonar.javascript.lcov.reportPaths=lattice/coverage/lcov.info needs: - - job: run go tests + - job: run go tests future - job: run jest tests build for linux amd64: @@ -210,6 +230,7 @@ build container fb: # 3. make sure docker/docker-compose is installed # 4. make sure the git config is done `git config --global --add url."ssh://git@github.com/".insteadOf "https://github.com/"` # 5. Add deploy key github.com/molecula/featurebase/settings/keys and add public key in .ssh folder of gitlab-runner user +# TODO: (I think) get clustertests coverage added to coverage report clustertests: stage: integration tags: @@ -220,6 +241,23 @@ clustertests: script: - make clustertests +external lookup tests: + stage: integration + image: golang:$GOVERSION + # TODO: no rules here, do we need to add the rules line? + variables: + POSTGRES_DB: $POSTGRES_DB + POSTGRES_USER: $POSTGRES_USER + POSTGRES_PASSWORD: $POSTGRES_PASSWORD + POSTGRES_HOST_AUTH_METHOD: trust + services: + - postgres:13.5 + script: + - apt-get update --allow-releaseinfo-change -y + - apt-get install -y postgresql-client + - go test . -run "^TestExternalLookup" -externalLookupDSN postgresql://$POSTGRES_USER:$POSTGRES_PASSWORD@postgres/$POSTGRES_DB?sslmode=disable + + smoke test: stage: integration image: registry.gitlab.com/gitlab-org/cloud-deploy/aws-base:latest diff --git a/client/client.go b/client/client.go index 211e7c3f7..082a56634 100644 --- a/client/client.go +++ b/client/client.go @@ -36,7 +36,7 @@ import ( const PQLVersion = "1.0" // DefaultShardWidth is used if an index doesn't have it defined. -const DefaultShardWidth = 1 << 20 +const DefaultShardWidth = pilosa.ShardWidth const maxHosts = 10 diff --git a/etcd/embed.go b/etcd/embed.go index 2e034eb67..db3bc34ea 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -227,12 +227,12 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) { break default: // nil, or an error we don't know about - return err + return errors.Wrap(err, "non-retryable error") } } // if we got here, we got a total of three of some combination of // ErrTimeout or ErrLeaderChanged, and we're giving up. - return err + return errors.Wrap(err, "exhausted all retries") } func parseOptions(opt Options) *embed.Config { diff --git a/http/handler.go b/http/handler.go index efd20d7e2..579f66385 100644 --- a/http/handler.go +++ b/http/handler.go @@ -54,7 +54,7 @@ type Handler struct { logger logger.Logger - querylogger logger.Logger + queryLogger logger.Logger // Keeps the query argument validators for each handler validators map[string]*queryValidationSpec @@ -152,7 +152,7 @@ func OptHandlerLogger(logger logger.Logger) handlerOption { func OptHandlerQueryLogger(logger logger.Logger) handlerOption { return func(h *Handler) error { - h.querylogger = logger + h.queryLogger = logger return nil } } @@ -285,13 +285,8 @@ const ( contextKeyQueryRequest contextKeyQuery = iota contextKeyQueryError contextKeyGroupMembership - contextKeyPermission ) -func GetContextKeyPermission() contextKeyQuery { - return contextKeyPermission -} - // addQueryContext puts the results of handler.readQueryRequest into the Context for use by // both other middleware and any handlers. func (h *Handler) addQueryContext(next http.Handler) http.Handler { @@ -433,20 +428,20 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/transaction/{id}", handler.chkAuthZ(handler.handlePostTransaction, authz.Read)).Methods("POST").Name("PostTransaction") router.HandleFunc("/transaction/{id}/finish", handler.chkAuthZ(handler.handlePostFinishTransaction, authz.Read)).Methods("POST").Name("PostFinishTransaction") router.HandleFunc("/transactions", handler.chkAuthZ(handler.handleGetTransactions, authz.Read)).Methods("GET").Name("GetTransactions") - router.HandleFunc("/queries", handler.chkAuthZ(handler.handleGetActiveQueries, authz.Read)).Methods("GET").Name("GetActiveQueries") - router.HandleFunc("/query-history", handler.chkAuthZ(handler.handleGetPastQueries, authz.Read)).Methods("GET").Name("GetPastQueries") - router.HandleFunc("/version", handler.chkAuthZ(handler.handleGetVersion, authz.Read)).Methods("GET").Name("GetVersion") + router.HandleFunc("/queries", handler.chkAuthZ(handler.handleGetActiveQueries, authz.Admin)).Methods("GET").Name("GetActiveQueries") + router.HandleFunc("/query-history", handler.chkAuthZ(handler.handleGetPastQueries, authz.Admin)).Methods("GET").Name("GetPastQueries") + router.HandleFunc("/version", handler.handleGetVersion).Methods("GET").Name("GetVersion") // /ui endpoints are for UI use; they may change at any time. router.HandleFunc("/ui/usage", handler.chkAuthZ(handler.handleGetUsage, authz.Read)).Methods("GET").Name("GetUsage") router.HandleFunc("/ui/transaction", handler.chkAuthZ(handler.handleGetTransactionList, authz.Read)).Methods("GET").Name("GetTransactionList") router.HandleFunc("/ui/transaction/", handler.chkAuthZ(handler.handleGetTransactionList, authz.Read)).Methods("GET").Name("GetTransactionList") - router.HandleFunc("/ui/shard-distribution", handler.chkAuthZ(handler.handleGetShardDistribution, authz.Read)).Methods("GET").Name("GetShardDistribution") + router.HandleFunc("/ui/shard-distribution", handler.chkAuthZ(handler.handleGetShardDistribution, authz.Admin)).Methods("GET").Name("GetShardDistribution") // /internal endpoints are for internal use only; they may change at any time. // DO NOT rely on these for external applications! - // Truly used internally by featurebease + // Truly used internally by featurebase router.HandleFunc("/internal/cluster/message", handler.chkInternal(handler.handlePostClusterMessage)).Methods("POST").Name("PostClusterMessage") router.HandleFunc("/internal/translate/data", handler.chkInternal(handler.handleGetTranslateData)).Methods("GET").Name("GetTranslateData") router.HandleFunc("/internal/translate/data", handler.chkInternal(handler.handlePostTranslateData)).Methods("POST").Name("PostTranslateData") @@ -459,23 +454,23 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/internal/partition/nodes", handler.chkAuthN(handler.handleGetPartitionNodes)).Methods("GET").Name("GetPartitionNodes") router.HandleFunc("/internal/translate/keys", handler.chkAuthN(handler.handlePostTranslateKeys)).Methods("POST").Name("PostTranslateKeys") router.HandleFunc("/internal/translate/ids", handler.chkAuthN(handler.handlePostTranslateIDs)).Methods("POST").Name("PostTranslateIDs") - router.HandleFunc("/internal/index/{index}/field/{field}/mutex-check", handler.chkAuthN(handler.handleInternalGetMutexCheck)).Methods("GET").Name("InternalGetMutexCheck") - router.HandleFunc("/internal/index/{index}/field/{field}/remote-available-shards/{shardID}", handler.chkAuthN(handler.handleDeleteRemoteAvailableShard)).Methods("DELETE") - router.HandleFunc("/internal/index/{index}/shard/{shard}/snapshot", handler.chkAuthN(handler.handleGetIndexShardSnapshot)).Methods("GET").Name("GetIndexShardSnapshot") - router.HandleFunc("/internal/index/{index}/shards", handler.chkAuthN(handler.handleGetIndexAvailableShards)).Methods("GET").Name("GetIndexAvailableShards") + router.HandleFunc("/internal/index/{index}/field/{field}/mutex-check", handler.chkAuthZ(handler.handleInternalGetMutexCheck, authz.Read)).Methods("GET").Name("InternalGetMutexCheck") + router.HandleFunc("/internal/index/{index}/field/{field}/remote-available-shards/{shardID}", handler.chkAuthZ(handler.handleDeleteRemoteAvailableShard, authz.Admin)).Methods("DELETE") + router.HandleFunc("/internal/index/{index}/shard/{shard}/snapshot", handler.chkAuthZ(handler.handleGetIndexShardSnapshot, authz.Read)).Methods("GET").Name("GetIndexShardSnapshot") + router.HandleFunc("/internal/index/{index}/shards", handler.chkAuthZ(handler.handleGetIndexAvailableShards, authz.Read)).Methods("GET").Name("GetIndexAvailableShards") router.HandleFunc("/internal/nodes", handler.chkAuthN(handler.handleGetNodes)).Methods("GET").Name("GetNodes") router.HandleFunc("/internal/shards/max", handler.chkAuthN(handler.handleGetShardsMax)).Methods("GET").Name("GetShardsMax") // TODO: deprecate, but it's being used by the client - router.HandleFunc("/internal/ingest/{index}", handler.chkAuthN(handler.handlePostIngestData)).Methods("POST").Name("PostIngestData") - router.HandleFunc("/internal/ingest/{index}/node", handler.chkAuthN(handler.handlePostIngestNode)).Methods("POST").Name("PostIngestNode") + router.HandleFunc("/internal/ingest/{index}", handler.chkAuthZ(handler.handlePostIngestData, authz.Write)).Methods("POST").Name("PostIngestData") + router.HandleFunc("/internal/ingest/{index}/node", handler.chkAuthZ(handler.handlePostIngestNode, authz.Write)).Methods("POST").Name("PostIngestNode") - router.HandleFunc("/internal/schema", handler.chkAuthN(handler.handleIngestSchema)).Methods("POST").Name("PostIngestSchema") - router.HandleFunc("/internal/translate/index/{index}/keys/find", handler.chkAuthN(handler.handleFindIndexKeys)).Methods("POST").Name("FindIndexKeys") - router.HandleFunc("/internal/translate/index/{index}/keys/create", handler.chkAuthN(handler.handleCreateIndexKeys)).Methods("POST").Name("CreateIndexKeys") - router.HandleFunc("/internal/translate/index/{index}/{partition}", handler.chkAuthN(handler.handlePostTranslateIndexDB)).Methods("POST").Name("PostTranslateIndexDB") - router.HandleFunc("/internal/translate/field/{index}/{field}", handler.chkAuthN(handler.handlePostTranslateFieldDB)).Methods("POST").Name("PostTranslateFieldDB") - router.HandleFunc("/internal/translate/field/{index}/{field}/keys/find", handler.chkAuthN(handler.handleFindFieldKeys)).Methods("POST").Name("FindFieldKeys") - router.HandleFunc("/internal/translate/field/{index}/{field}/keys/create", handler.chkAuthN(handler.handleCreateFieldKeys)).Methods("POST").Name("CreateFieldKeys") - router.HandleFunc("/internal/translate/field/{index}/{field}/keys/like", handler.chkAuthN(handler.handleMatchField)).Methods("POST").Name("MatchFieldKeys") + router.HandleFunc("/internal/schema", handler.chkAuthZ(handler.handleIngestSchema, authz.Admin)).Methods("POST").Name("PostIngestSchema") + router.HandleFunc("/internal/translate/index/{index}/keys/find", handler.chkAuthZ(handler.handleFindIndexKeys, authz.Admin)).Methods("POST").Name("FindIndexKeys") + router.HandleFunc("/internal/translate/index/{index}/keys/create", handler.chkAuthZ(handler.handleCreateIndexKeys, authz.Admin)).Methods("POST").Name("CreateIndexKeys") + router.HandleFunc("/internal/translate/index/{index}/{partition}", handler.chkAuthZ(handler.handlePostTranslateIndexDB, authz.Admin)).Methods("POST").Name("PostTranslateIndexDB") + router.HandleFunc("/internal/translate/field/{index}/{field}", handler.chkAuthZ(handler.handlePostTranslateFieldDB, authz.Admin)).Methods("POST").Name("PostTranslateFieldDB") + router.HandleFunc("/internal/translate/field/{index}/{field}/keys/find", handler.chkAuthZ(handler.handleFindFieldKeys, authz.Admin)).Methods("POST").Name("FindFieldKeys") + router.HandleFunc("/internal/translate/field/{index}/{field}/keys/create", handler.chkAuthZ(handler.handleCreateFieldKeys, authz.Admin)).Methods("POST").Name("CreateFieldKeys") + router.HandleFunc("/internal/translate/field/{index}/{field}/keys/like", handler.chkAuthZ(handler.handleMatchField, authz.Read)).Methods("POST").Name("MatchFieldKeys") router.HandleFunc("/internal/idalloc/reserve", handler.chkAuthN(handler.handleReserveIDs)).Methods("POST").Name("ReserveIDs") router.HandleFunc("/internal/idalloc/commit", handler.chkAuthN(handler.handleCommitIDs)).Methods("POST").Name("CommitIDs") @@ -483,9 +478,9 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/internal/idalloc/reset/{index}", handler.chkAuthN(handler.handleResetIDAlloc)).Methods("POST").Name("ResetIDAlloc") router.HandleFunc("/internal/idalloc/data", handler.chkAuthN(handler.handleIDAllocData)).Methods("GET").Name("IDAllocData") - router.HandleFunc("/internal/restore/{index}/{shardID}", handler.chkAuthN(handler.handlePostRestore)).Methods("POST").Name("Restore") + router.HandleFunc("/internal/restore/{index}/{shardID}", handler.chkAuthZ(handler.handlePostRestore, authz.Admin)).Methods("POST").Name("Restore") - router.HandleFunc("/internal/debug/rbf", handler.chkAuthN(handler.handleGetInternalDebugRBFJSON)).Methods("GET").Name("GetInternalDebugRBFJSON") + router.HandleFunc("/internal/debug/rbf", handler.chkAuthZ(handler.handleGetInternalDebugRBFJSON, authz.Admin)).Methods("GET").Name("GetInternalDebugRBFJSON") // endpoints for collecting cpu profiles from a chosen begin point to // when the client wants to stop. Used for profiling imports that @@ -573,76 +568,112 @@ func (h *Handler) chkAuthN(handler http.HandlerFunc) http.HandlerFunc { func (h *Handler) chkAuthZ(handler http.HandlerFunc, perm authz.Permission) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { - lperm := perm - if h.auth != nil { - uinfo, err := h.auth.Authenticate(getToken(r)) - if err != nil { - http.Error(w, errors.Wrap(err, "authenticating").Error(), http.StatusForbidden) - return - } - - if h.permissions == nil { - h.logger.Errorf("authentication is turned on without authorization permissions set") - http.Error(w, "authorizing", http.StatusInternalServerError) - return - } - - ctx := context.WithValue(r.Context(), contextKeyGroupMembership, uinfo.Groups) - - if h.permissions.IsAdmin(uinfo.Groups) { - ctx = context.WithValue(ctx, contextKeyPermission, authz.Admin) - handler.ServeHTTP(w, r.WithContext(ctx)) - return - } else if lperm == authz.Admin { - http.Error(w, "Insufficient permissions: user does not have admin permission", http.StatusForbidden) - return - } - - var queryString string - queryRequest := r.Context().Value(contextKeyQueryRequest) - if req, ok := queryRequest.(*pilosa.QueryRequest); ok { - queryString = req.Query - - q, err := pql.ParseString(queryString) - if err != nil { - http.Error(w, errors.Wrap(err, "parsing query string").Error(), http.StatusBadRequest) - return - } - if q.WriteCallN() > 0 { - lperm = authz.Write - } - } - - queryString = strings.Replace(queryString, "\n", "", -1) - - if r.Method == "POST" { - h.querylogger.Infof("User ID: %s, User Name: %s, Endpoint: %s, Index: %s, Query: %s, Err: %v", uinfo.UserID, uinfo.UserName, r.URL.Path, "indexName", queryString, err) - } - - indexName, ok := mux.Vars(r)["index"] - if ok { - p, err := h.permissions.GetPermissions(uinfo, indexName) - ctx = context.WithValue(ctx, contextKeyPermission, p) - if err != nil { - w.Header().Add("Content-Type", "text/plain") - http.Error(w, errors.Wrap(err, "Insufficient Permissions").Error(), http.StatusForbidden) - return - } - if !p.Satisfies(lperm) { - w.Header().Add("Content-Type", "text/plain") - http.Error(w, fmt.Sprintf("Insufficient permissions: user has %s permissions, but request requires %s permission", p, lperm), http.StatusForbidden) - return - } - } - - handler.ServeHTTP(w, r.WithContext(ctx)) - } else { + // if auth isn't turned on, just serve the request + if h.auth == nil { handler.ServeHTTP(w, r) + return } + // make a copy of the requested permissions + lperm := perm + + // check if the user is authenticated + uinfo, err := h.auth.Authenticate(getToken(r)) + if err != nil { + http.Error(w, errors.Wrap(err, "authenticating").Error(), http.StatusForbidden) + return + } + + // put the user's groups in the context + ctx := context.WithValue(r.Context(), contextKeyGroupMembership, uinfo.Groups) + + // unlikely h.permissions will be nil, but we'll check to be safe + if h.permissions == nil { + h.logger.Errorf("authentication is turned on without authorization permissions set") + http.Error(w, "authorizing", http.StatusInternalServerError) + return + } + + // figure out what the user is querying for + queryString := "" + queryRequest := r.Context().Value(contextKeyQueryRequest) + if req, ok := queryRequest.(*pilosa.QueryRequest); ok { + queryString = req.Query + + q, err := pql.ParseString(queryString) + if err != nil { + http.Error(w, errors.Wrap(err, "parsing query string").Error(), http.StatusBadRequest) + return + } + + // if there are write calls, and the needed perms don't already + // satisfy write permissions, then make them write permissions + if q.WriteCallN() > 0 && !lperm.Satisfies(authz.Write) { + lperm = authz.Write + } + } + // make the query string pretty + queryString = strings.Replace(queryString, "\n", "", -1) + + // figure out if we should log this query + toLog := true + for _, ep := range []string{"/status", "/metrics", "/info", "/internal"} { + if strings.HasPrefix(r.URL.Path, ep) { + toLog = false + break + } + } + if toLog { + h.queryLogger.Infof("%v, %v, %v, %v, %v, %v", GetIP(r), r.UserAgent(), r.URL.Path, uinfo.UserID, uinfo.UserName, queryString) + } + + // if they're an admin, they can do whatever they want + if h.permissions.IsAdmin(uinfo.Groups) { + handler.ServeHTTP(w, r.WithContext(ctx)) + return + } else if lperm == authz.Admin { + // if they're not an admin, and they need to be, we can just + // error right here + http.Error(w, "Insufficient permissions: user does not have admin permission", http.StatusForbidden) + return + } + + // try to get the index name + indexName, ok := mux.Vars(r)["index"] + if !ok { + indexName = r.URL.Query().Get("index") + } + + // if we have an index name, then we check the user permissions + // against that index + if indexName != "" { + p, err := h.permissions.GetPermissions(uinfo, indexName) + if err != nil { + w.Header().Add("Content-Type", "text/plain") + http.Error(w, errors.Wrap(err, "Insufficient Permissions").Error(), http.StatusForbidden) + return + } + + // if they're not permitted to access this index, error + if !p.Satisfies(lperm) { + w.Header().Add("Content-Type", "text/plain") + http.Error(w, "Insufficient permissions", http.StatusForbidden) + return + } + } + handler.ServeHTTP(w, r.WithContext(ctx)) + } } +func GetIP(r *http.Request) string { + forwarded := r.Header.Get("X-FORWARDED-FOR") + if forwarded != "" { + return forwarded + } + return r.RemoteAddr +} + // statikHandler implements the http.Handler interface, and responds to // requests for static assets with the appropriate file contents embedded // in a statik filesystem. @@ -811,30 +842,6 @@ func headerAcceptRoaringRow(header http.Header) bool { return false } -func (h *Handler) filterResponse(w http.ResponseWriter, r *http.Request, schema []*pilosa.IndexInfo) []*pilosa.IndexInfo { - if h.auth != nil { - g := r.Context().Value(contextKeyGroupMembership) - if g == nil { - http.Error(w, "Forbidden", http.StatusForbidden) - return nil - } - indexes := h.permissions.GetAuthorizedIndexList(g.([]authn.Group), authz.Read) - var new []*pilosa.IndexInfo - for _, s := range schema { - for _, index := range indexes { - if s.Name == index { - new = append(new, s) - } - } - - } - return new - - } - return schema - -} - // handleGetSchema handles GET /schema requests. func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { @@ -851,7 +858,27 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) { h.logger.Printf("getting schema error: %s", err) } - schema = h.filterResponse(w, r, schema) + // if auth is turned on, filter response to only include authorized indexes + if h.auth != nil { + g := r.Context().Value(contextKeyGroupMembership) + if g == nil { + http.Error(w, "Forbidden", http.StatusForbidden) + return + } + if !h.permissions.IsAdmin(g.([]authn.Group)) { + var filtered []*pilosa.IndexInfo + allowed := h.permissions.GetAuthorizedIndexList(g.([]authn.Group), authz.Read) + for _, s := range schema { + for _, index := range allowed { + if s.Name == index { + filtered = append(filtered, s) + break + } + } + } + schema = filtered + } + } if err := json.NewEncoder(w).Encode(pilosa.Schema{Indexes: schema}); err != nil { h.logger.Errorf("write schema response error: %s", err) @@ -871,7 +898,28 @@ func (h *Handler) handleGetSchemaDetails(w http.ResponseWriter, r *http.Request) h.logger.Printf("error getting detailed schema: %s", err) return } - schema = h.filterResponse(w, r, schema) + + // if auth is turned on, filter response to only include authorized indexes + if h.auth != nil { + g := r.Context().Value(contextKeyGroupMembership) + if g == nil { + http.Error(w, "Forbidden", http.StatusForbidden) + return + } + if !h.permissions.IsAdmin(g.([]authn.Group)) { + var filtered []*pilosa.IndexInfo + allowed := h.permissions.GetAuthorizedIndexList(g.([]authn.Group), authz.Read) + for _, s := range schema { + for _, index := range allowed { + if s.Name == index { + filtered = append(filtered, s) + break + } + } + } + schema = filtered + } + } if err := json.NewEncoder(w).Encode(pilosa.Schema{Indexes: schema}); err != nil { h.logger.Printf("write schema response error: %s", err) } @@ -917,6 +965,38 @@ func (h *Handler) handleGetUsage(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusInternalServerError) } + // if auth is turned on, filter results + if h.auth != nil { + g := r.Context().Value(contextKeyGroupMembership) + if g == nil { + http.Error(w, "Forbidden", http.StatusForbidden) + return + } + if !h.permissions.IsAdmin(g.([]authn.Group)) { + allowed := h.permissions.GetAuthorizedIndexList(g.([]authn.Group), authz.Read) + filteredNodeUsages := map[string]pilosa.NodeUsage{} + + for nodeId, nodeUsage := range nodeUsages { + filteredIndexUsage := pilosa.NodeUsage{ + Disk: pilosa.DiskUsage{ + IndexUsage: map[string]pilosa.IndexUsage{}, + }, + } + for index, idxUsage := range nodeUsage.Disk.IndexUsage { + // is it in auth list + for _, authd := range allowed { + if index == authd { + filteredIndexUsage.Disk.IndexUsage[index] = idxUsage + break + } + } + } + filteredNodeUsages[nodeId] = filteredIndexUsage + } + nodeUsages = filteredNodeUsages + } + } + w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(nodeUsages); err != nil { h.logger.Errorf("write status response error: %s", err) diff --git a/http/handler_internal_test.go b/http/handler_internal_test.go index e23af836c..511079961 100644 --- a/http/handler_internal_test.go +++ b/http/handler_internal_test.go @@ -220,7 +220,7 @@ func TestAuthentication(t *testing.T) { h := Handler{ logger: logger.NewStandardLogger(os.Stdout), - querylogger: logger.NewStandardLogger(os.Stdout), + queryLogger: logger.NewStandardLogger(os.Stdout), auth: a, } diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 806d68101..79d067ad4 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -4,6 +4,7 @@ package clustertest import ( "context" "fmt" + "io" "net/http" "os" "os/exec" @@ -124,7 +125,11 @@ func TestClusterStuff(t *testing.T) { } else if resp, err := client.Do(req); err != nil { t.Fatalf("doing request: %v", err) } else if resp.StatusCode >= 400 { - t.Fatalf("bad response: %v", resp) + bod, readErr := io.ReadAll(resp.Body) + if readErr != nil { + t.Logf("reading error body: %v", readErr) + } + t.Fatalf("deleting index: code=%d, body=%s", resp.StatusCode, bod) } var restoreCmd *exec.Cmd diff --git a/lattice/Dockerfile b/lattice/Dockerfile index 2b30d6cfb..6aa669e54 100644 --- a/lattice/Dockerfile +++ b/lattice/Dockerfile @@ -1,10 +1,10 @@ FROM moleculacorp/nodejs:latest as build - +# make sure that your docker settings allow for at least like 4gb of ram, it +# takes a lot to build this WORKDIR /lattice - COPY package.json ./ -COPY yarn.lock ./ -RUN yarn install +RUN apk update && apk upgrade yarn +RUN yarn install --network-timeout 100000 COPY . ./ RUN yarn build diff --git a/qa/simulacraData/simulacra_data_test.go b/qa/simulacraData/simulacra_data_test.go index ae16de547..c5d7e59ae 100644 --- a/qa/simulacraData/simulacra_data_test.go +++ b/qa/simulacraData/simulacra_data_test.go @@ -2,12 +2,14 @@ package main import ( + "os" "testing" ) const testRecords int = 1000 func TestAge(t *testing.T) { + defer os.Remove("age.csv") err := GenerateAgeField(testRecords) if err != nil { t.Fatalf("%v", err) @@ -15,13 +17,15 @@ func TestAge(t *testing.T) { } func TestIP(t *testing.T) { + defer os.Remove("ip.csv") err := GenerateIPField(testRecords) if err != nil { t.Fatalf("%v", err) } } -func TestIndentifer(t *testing.T) { +func TestIdentifier(t *testing.T) { + defer os.Remove("identifier.csv") err := GenerateArbIdField(testRecords) if err != nil { t.Fatalf("%v", err) @@ -29,6 +33,7 @@ func TestIndentifer(t *testing.T) { } func TestOptIn(t *testing.T) { + defer os.Remove("optin.csv") err := GenerateOptInField(testRecords) if err != nil { t.Fatalf("%v", err) @@ -36,6 +41,7 @@ func TestOptIn(t *testing.T) { } func TestCountry(t *testing.T) { + defer os.Remove("country.csv") err := GenerateCountryField(testRecords) if err != nil { t.Fatalf("%v", err) @@ -43,6 +49,7 @@ func TestCountry(t *testing.T) { } func TestTime(t *testing.T) { + defer os.Remove("time.csv") err := GenerateTimeField(testRecords) if err != nil { t.Fatalf("%v", err) diff --git a/server.go b/server.go index 351e3818c..23e760e19 100644 --- a/server.go +++ b/server.go @@ -67,7 +67,7 @@ type Server struct { // nolint: maligned systemInfo SystemInfo gcNotifier GCNotifier logger logger.Logger - querylogger logger.Logger + queryLogger logger.Logger snapshotQueue SnapshotQueue nodeID string @@ -115,7 +115,7 @@ func OptServerLogger(l logger.Logger) ServerOption { func OptServerQueryLogger(l logger.Logger) ServerOption { return func(s *Server) error { - s.querylogger = l + s.queryLogger = l return nil } } diff --git a/server/grpc.go b/server/grpc.go index 3683de99a..f24823915 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -27,6 +27,7 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/peer" "google.golang.org/grpc/reflection" "google.golang.org/grpc/status" ) @@ -36,6 +37,7 @@ type GRPCHandler struct { api *pilosa.API perms *authz.GroupPermissions logger logger.Logger + queryLogger logger.Logger stats stats.StatsClient inspectDeprecated sync.Once } @@ -59,6 +61,11 @@ func (h *GRPCHandler) WithPerms(perms *authz.GroupPermissions) *GRPCHandler { return h } +func (h *GRPCHandler) WithQueryLogger(logger logger.Logger) *GRPCHandler { + h.queryLogger = logger + return h +} + // errorToStatusError appends an appropriate grpc status code // to the error (returning it as a status.Error). func errToStatusError(err error) error { @@ -166,15 +173,18 @@ func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQ if err != nil { return errors.Wrap(err, "parsing SQL") } + + allowed := h.perms.GetAuthorizedIndexList(uinfo.(*authn.UserInfo).Groups, authz.Read) if !h.perms.IsAdmin(uinfo.(*authn.UserInfo).Groups) { - if !isAllowed(parsed.Tables, h.perms.GetAuthorizedIndexList(uinfo.(*authn.UserInfo).Groups, authz.Read)) { + if !isAllowed(parsed.Tables, allowed) { return status.Error(codes.PermissionDenied, "insufficient permissions to access requested tables") } + ctx = context.WithValue(ctx, "indices", allowed) } } start := time.Now() - results, err := h.execSQL(stream.Context(), req.Sql) + results, err := h.execSQL(ctx, req.Sql) duration := time.Since(start) if err != nil { return err @@ -208,6 +218,10 @@ func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQ // https://github.com/molecula/pilosa/pull/644 func (h *GRPCHandler) QuerySQLUnary(ctx context.Context, req *pb.QuerySQLRequest) (*pb.TableResponse, error) { start := time.Now() + uinfo := ctx.Value("userinfo") + if uinfo != nil && !h.perms.IsAdmin(uinfo.(*authn.UserInfo).Groups) { + ctx = context.WithValue(ctx, "indices", h.perms.GetAuthorizedIndexList(uinfo.(*authn.UserInfo).Groups, authz.Read)) + } results, err := h.execSQL(ctx, req.Sql) if err != nil { return nil, err @@ -1425,8 +1439,9 @@ type grpcServer struct { auth *authn.Auth perms *authz.GroupPermissions - logger logger.Logger - stats stats.StatsClient + logger logger.Logger + queryLogger logger.Logger + stats stats.StatsClient } type grpcServerOption func(s *grpcServer) error @@ -1480,6 +1495,13 @@ func OptGRPCServerPerm(gp *authz.GroupPermissions) grpcServerOption { } } +func OptGRPCServerQueryLogger(logger logger.Logger) grpcServerOption { + return func(s *grpcServer) error { + s.queryLogger = logger + return nil + } +} + func (s *grpcServer) Serve() error { s.logger.Infof("enabled grpc listening on %s", s.ln.Addr()) @@ -1539,7 +1561,7 @@ func NewGRPCServer(opts ...grpcServerOption) (*grpcServer, error) { if server.auth != nil { gopts = append(gopts, grpc.UnaryInterceptor( func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { - ctx, err := Valid(ctx, server.auth) + ctx, err := Valid(ctx, info.FullMethod, server.auth, req, server.queryLogger) if err != nil { return nil, err } @@ -1548,18 +1570,18 @@ func NewGRPCServer(opts ...grpcServerOption) (*grpcServer, error) { )) gopts = append(gopts, grpc.StreamInterceptor( func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { - ctx, err := Valid(ss.Context(), server.auth) + ctx, err := Valid(ss.Context(), info.FullMethod, server.auth, srv, server.queryLogger) if err != nil { return err } - return handler(srv, newWrappedStream(ss, ctx)) + return handler(srv, &wrappedStream{ss, ctx}) }, )) } // create grpc server server.grpcServer = grpc.NewServer(gopts...) - grpcHandler := NewGRPCHandler(server.api).WithLogger(server.logger).WithStats(server.stats) + grpcHandler := NewGRPCHandler(server.api).WithLogger(server.logger).WithStats(server.stats).WithQueryLogger(server.queryLogger) // add server permissions if we've got 'em if server.perms != nil { @@ -1585,6 +1607,7 @@ type wrappedStream struct { func (w *wrappedStream) Context() context.Context { return w.uiContext } + func (w *wrappedStream) RecvMsg(m interface{}) error { return w.ServerStream.RecvMsg(m) } @@ -1593,39 +1616,46 @@ func (w *wrappedStream) SendMsg(m interface{}) error { return w.ServerStream.SendMsg(m) } -func newWrappedStream(s grpc.ServerStream, ctx context.Context) grpc.ServerStream { - return &wrappedStream{s, ctx} -} - -func Valid(ctx context.Context, auth *authn.Auth) (context.Context, error) { +func Valid(ctx context.Context, method string, auth *authn.Auth, req interface{}, logger logger.Logger) (context.Context, error) { md, ok := metadata.FromIncomingContext(ctx) if !ok { return ctx, status.Errorf(codes.InvalidArgument, "missing metadata") } - authorization, ok := md["authorization"] + authorization, ok := md["authorization"] if !ok { - c, ok := md["cookie"] - if !ok { + c, there := md["cookie"] + if !there { return ctx, status.Errorf(codes.InvalidArgument, "missing authorization token") } cookies := strings.Split(c[0], "; ") for _, cookie := range cookies { if strings.HasPrefix(cookie, "molecula-chip") { authorization = strings.Split(cookie, "molecula-chip=")[1:] + break } } - if len(authorization) == 0 { - return ctx, status.Errorf(codes.InvalidArgument, "missing authorization token") - - } + } + if len(authorization) == 0 { + return ctx, status.Errorf(codes.InvalidArgument, "missing authorization token") } token := strings.TrimPrefix(authorization[0], "Bearer ") - userinfo, err := auth.Authenticate(token) + uinfo, err := auth.Authenticate(token) if err != nil { return ctx, status.Errorf(codes.Unauthenticated, err.Error()) } - return context.WithValue(ctx, "userinfo", userinfo), nil + p, ok := peer.FromContext(ctx) + ip := "" + if ok { + ip = p.Addr.String() + } + ua, ok := md["user-agent"] + if !ok { + ua = []string{""} + } + logger.Infof("GRPC: %v, %v, %v, %v, %v, %v", ip, ua, method, uinfo.UserID, uinfo.UserName, req) + + return context.WithValue(ctx, "userinfo", uinfo), nil } diff --git a/server/grpc_test.go b/server/grpc_test.go index 63482f687..c2b993a96 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -18,6 +18,7 @@ import ( pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/authn" "github.com/molecula/featurebase/v2/authz" + "github.com/molecula/featurebase/v2/logger" "github.com/molecula/featurebase/v2/pql" pb "github.com/molecula/featurebase/v2/proto" "github.com/molecula/featurebase/v2/server" @@ -970,7 +971,7 @@ func TestQuerySQL(t *testing.T) { } } -func TestQuerySQLUnaryWithError(t *testing.T) { +func TestQuerySQLWithError(t *testing.T) { stream := &MockServerTransportStream{} ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream) @@ -1060,14 +1061,13 @@ admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` } validToken = "Bearer " + validToken - adminUser := &authn.UserInfo{ + return &authn.UserInfo{ UserID: "fake" + name, UserName: name, Groups: groups, Token: validToken, Expiry: time.Time{}, } - return adminUser } user := makeUser([]authn.Group{{GroupID: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", GroupName: "adminGroup"}}, "admin") @@ -1076,7 +1076,7 @@ admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` "userinfo", user, ) - readuser := makeUser([]authn.Group{{GroupID: "dca35310-ecda-4f23-86cd-876aee55906b", GroupName: "readers"}}, "admin") + readuser := makeUser([]authn.Group{{GroupID: "dca35310-ecda-4f23-86cd-876aee55906b", GroupName: "readers"}}, "reader") readCtx := context.WithValue( ctx, "userinfo", @@ -1110,6 +1110,16 @@ admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` t.Fatal(err) } }) + + t.Run("test-admin-auth-sql-show", func(t *testing.T) { + mock := &mockPilosa_QuerySQLServer{ctx: adminCtx} + + err := gh.QuerySQL(&pb.QuerySQLRequest{Sql: "show tables"}, mock) + if err != nil { + t.Fatal(err) + } + }) + t.Run("test-admin-auth-get-index", func(t *testing.T) { _, err := gh.GetIndex(adminCtx, &pb.GetIndexRequest{Name: "grouper"}) if err != nil { @@ -1143,6 +1153,23 @@ admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` t.Fatal(err) } }) + t.Run("test-show-tables-unary", func(t *testing.T) { + response, err := gh.QuerySQLUnary(readCtx, &pb.QuerySQLRequest{ + Sql: "show tables", + }) + + if err != nil && len(response.Rows) != 1 { + t.Fatal(err) + } + }) + t.Run("test-show-tables-unary-admin", func(t *testing.T) { + response, err := gh.QuerySQLUnary(adminCtx, &pb.QuerySQLRequest{ + Sql: "show tables", + }) + if err != nil && len(response.Rows) != 3 { + t.Fatal(err) + } + }) t.Run("test-write-with-write-auth-pql", func(t *testing.T) { _, err := gh.QueryPQLUnary(writeCtx, &pb.QueryPQLRequest{ Index: "grouper", @@ -1349,8 +1376,7 @@ func setUpTestQuerySQLUnary(ctx context.Context, t *testing.T) (gh *server.GRPCH t.Helper() m := test.RunCommand(t) - gh = server.NewGRPCHandler(m.API) - + gh = server.NewGRPCHandler(m.API).WithQueryLogger(logger.NewStandardLogger(os.Stdout)) // grouper grouper := m.MustCreateIndex(t, "grouper", pilosa.IndexOptions{Keys: false, TrackExistence: true}) m.MustCreateField(t, grouper.Name(), "color", pilosa.OptFieldKeys()) diff --git a/server/server.go b/server/server.go index c2dc433cd..781a1a216 100644 --- a/server/server.go +++ b/server/server.go @@ -70,9 +70,9 @@ type Command struct { done chan struct{} logOutput io.Writer - querylogOutput io.Writer + queryLogOutput io.Writer logger loggerLogger - querylogger loggerLogger + queryLogger loggerLogger Handler pilosa.Handler grpcServer *grpcServer @@ -476,7 +476,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(c, &sync.Mutex{})), pilosa.OptServerOpenIDAllocator(pilosa.OpenIDAllocator), pilosa.OptServerLogger(m.logger), - pilosa.OptServerQueryLogger(m.querylogger), + pilosa.OptServerQueryLogger(m.queryLogger), pilosa.OptServerSystemInfo(gopsutil.NewSystemInfo()), pilosa.OptServerGCNotifier(gcnotify.NewActiveGCNotifier()), pilosa.OptServerStatsClient(statsClient), @@ -545,11 +545,12 @@ func (m *Command) SetupServer() error { err = m.setupQueryLogger() if err != nil { - return errors.Wrap(err, "setting up querylogger") + return errors.Wrap(err, "setting up queryLogger") } - m.querylogger.Infof("Group with admin level access: %v", p.Admin) - m.querylogger.Infof("Permissions: %+v", p.Permissions) + m.queryLogger.Infof("Featurebase Server Started") + m.queryLogger.Infof("Group with admin level access: %v", p.Admin) + m.queryLogger.Infof("Permissions: %+v", p.Permissions) // disable postgres binding if auth is enabled m.Config.Postgres.Bind = "" @@ -569,13 +570,14 @@ func (m *Command) SetupServer() error { OptGRPCServerStats(statsClient), OptGRPCServerAuth(m.auth), OptGRPCServerPerm(&p), + OptGRPCServerQueryLogger(m.queryLogger), ) m.Handler, err = http.NewHandler( http.OptHandlerAllowedOrigins(m.Config.Handler.AllowedOrigins), http.OptHandlerAPI(m.API), http.OptHandlerLogger(m.logger), - http.OptHandlerQueryLogger(m.querylogger), + http.OptHandlerQueryLogger(m.queryLogger), http.OptHandlerFileSystem(&statik.FileSystem{}), http.OptHandlerListener(m.ln, m.Config.Advertise), http.OptHandlerCloseTimeout(m.closeTimeout), @@ -642,16 +644,16 @@ func (m *Command) setupQueryLogger() error { return errors.Wrap(err, "opening file") } } - m.querylogOutput = f + m.queryLogOutput = f - m.querylogger = logger.NewStandardLogger(m.querylogOutput) + m.queryLogger = logger.NewStandardLogger(m.queryLogOutput) sighup := make(chan os.Signal, 1) signal.Notify(sighup, syscall.SIGHUP) go func() { for range sighup { if err := f.Reopen(); err != nil { - m.querylogger.Infof("reopen: %s\n", err.Error()) + m.queryLogger.Infof("reopen: %s\n", err.Error()) } } }() diff --git a/sql/show.go b/sql/show.go index 2848a77d7..186d51923 100644 --- a/sql/show.go +++ b/sql/show.go @@ -5,9 +5,11 @@ import ( "context" "fmt" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" pproto "github.com/molecula/featurebase/v2/proto" "github.com/pkg/errors" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "vitess.io/vitess/go/vt/sqlparser" ) @@ -46,16 +48,32 @@ func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Sh return nil, errors.Wrap(err, "getting schema") } - result := make(pproto.ConstRowser, len(indexInfo)) - for i, ii := range indexInfo { - result[i] = pproto.RowResponse{ + allowed, ok := ctx.Value("indices").([]string) + + result := make(pproto.ConstRowser, 0) + for _, ii := range indexInfo { + if ok { + // if authorization is turned on, allowed will be a list + // so we have to check if the index is in the allowed list + found := false + for _, idx := range allowed { + if ii.Name == idx { + found = true + break + } + } + if !found { + continue + } + } + result = append(result, pproto.RowResponse{ Headers: []*pproto.ColumnInfo{ {Name: "Table", Datatype: "string"}, }, Columns: []*pproto.ColumnResponse{ {ColumnVal: &pproto.ColumnResponse_StringVal{StringVal: ii.Name}}, }, - } + }) } // Sort the result. @@ -64,6 +82,19 @@ func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Sh func (s *ShowHandler) execShowFields(ctx context.Context, showStmt *sqlparser.Show) (pproto.ToRowser, error) { indexName := showStmt.OnTable.ToViewName().Name.String() + allowed, ok := ctx.Value("indices").([]string) + if ok { + found := false + for _, idx := range allowed { + if idx == indexName { + found = true + break + } + } + if !found { + return nil, status.Error(codes.PermissionDenied, "insufficient permissions to access requested tables") + } + } index, err := s.api.Index(ctx, indexName) if err != nil { return nil, errors.Wrap(err, "getting schema")