mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-10-06 19:07:50 +00:00
Merge branch 'master' into ui-fixes
This commit is contained in:
commit
580a6109fb
14 changed files with 401 additions and 181 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -20,3 +20,4 @@ launch.json
|
|||
__pycache__/
|
||||
report.xml
|
||||
outputs.json
|
||||
builds/
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
316
http/handler.go
316
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)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
|
|
|||
41
sql/show.go
41
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")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue