Add namespace filtering

pkg/client/factory.go

@@ -14,11 +14,13 @@ import (
 )
 
 type Factory struct {
-	impersonate    bool
+	impersonate         bool
-	clientCfg      *rest.Config
+	tableClientCfg      *rest.Config
-	watchClientCfg *rest.Config
+	tableWatchClientCfg *rest.Config
-	metadata       metadata.Interface
+	clientCfg           *rest.Config
-	Config         *rest.Config
+	watchClientCfg      *rest.Config
+	metadata            metadata.Interface
+	Config              *rest.Config
 }
 
 type addQuery struct {
@@ -41,17 +43,23 @@ func NewFactory(cfg *rest.Config, impersonate bool) (*Factory, error) {
 	clientCfg.QPS = 10000
 	clientCfg.Burst = 100
 	clientCfg.AcceptContentTypes = "application/json;as=Table;v=v1;g=meta.k8s.io"
-	clientCfg.Wrap(func(rt http.RoundTripper) http.RoundTripper {
+
+	watchClientCfg := rest.CopyConfig(clientCfg)
+	watchClientCfg.Timeout = 30 * time.Minute
+
+	setTable := func(rt http.RoundTripper) http.RoundTripper {
 		return &addQuery{
 			values: map[string]string{
 				"includeObject": "Object",
 			},
 			next: rt,
 		}
-	})
+	}
 
-	watchClientCfg := rest.CopyConfig(clientCfg)
+	tableClientCfg := rest.CopyConfig(clientCfg)
-	watchClientCfg.Timeout = 30 * time.Minute
+	tableClientCfg.Wrap(setTable)
+	tableWatchClientCfg := rest.CopyConfig(watchClientCfg)
+	tableWatchClientCfg.Wrap(setTable)
 
 	md, err := metadata.NewForConfig(cfg)
 	if err != nil {
@@ -59,11 +67,13 @@ func NewFactory(cfg *rest.Config, impersonate bool) (*Factory, error) {
 	}
 
 	return &Factory{
-		metadata:       md,
+		metadata:            md,
-		impersonate:    impersonate,
+		impersonate:         impersonate,
-		clientCfg:      clientCfg,
+		tableClientCfg:      tableClientCfg,
-		watchClientCfg: watchClientCfg,
+		tableWatchClientCfg: tableWatchClientCfg,
-		Config:         watchClientCfg,
+		clientCfg:           clientCfg,
+		watchClientCfg:      watchClientCfg,
+		Config:              watchClientCfg,
 	}, nil
 }
 
@@ -79,12 +89,20 @@ func (p *Factory) AdminClient(ctx *types.APIRequest, s *types.APISchema, namespa
 	return newClient(ctx, p.clientCfg, s, namespace, false)
 }
 
-func (p *Factory) ClientForWatch(ctx *types.APIRequest, s *types.APISchema, namespace string) (dynamic.ResourceInterface, error) {
+func (p *Factory) TableClient(ctx *types.APIRequest, s *types.APISchema, namespace string) (dynamic.ResourceInterface, error) {
-	return newClient(ctx, p.watchClientCfg, s, namespace, p.impersonate)
+	return newClient(ctx, p.tableClientCfg, s, namespace, p.impersonate)
 }
 
-func (p *Factory) AdminClientForWatch(ctx *types.APIRequest, s *types.APISchema, namespace string) (dynamic.ResourceInterface, error) {
+func (p *Factory) TableAdminClient(ctx *types.APIRequest, s *types.APISchema, namespace string) (dynamic.ResourceInterface, error) {
-	return newClient(ctx, p.watchClientCfg, s, namespace, false)
+	return newClient(ctx, p.tableClientCfg, s, namespace, false)
+}
+
+func (p *Factory) TableClientForWatch(ctx *types.APIRequest, s *types.APISchema, namespace string) (dynamic.ResourceInterface, error) {
+	return newClient(ctx, p.tableWatchClientCfg, s, namespace, p.impersonate)
+}
+
+func (p *Factory) TableAdminClientForWatch(ctx *types.APIRequest, s *types.APISchema, namespace string) (dynamic.ResourceInterface, error) {
+	return newClient(ctx, p.tableWatchClientCfg, s, namespace, false)
 }
 
 func newClient(ctx *types.APIRequest, cfg *rest.Config, s *types.APISchema, namespace string, impersonate bool) (dynamic.ResourceInterface, error) {

pkg/schema/collection.go

@@ -2,17 +2,20 @@ package schema
 
 import (
 	"context"
+	"net/http"
 	"strings"
 	"sync"
 
 	"github.com/rancher/steve/pkg/accesscontrol"
 	"github.com/rancher/steve/pkg/attributes"
+	"github.com/rancher/steve/pkg/schemaserver/server"
 	"github.com/rancher/steve/pkg/schemaserver/types"
 	"github.com/rancher/wrangler/pkg/name"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/cache"
 	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/endpoints/request"
 )
 
 type Factory interface {
@@ -47,6 +50,28 @@ type Template struct {
 	StoreFactory func(types.Store) types.Store
 }
 
+func WrapServer(factory Factory, server *server.Server) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		user, ok := request.UserFrom(req.Context())
+		if !ok {
+			return
+		}
+
+		schemas, err := factory.Schemas(user)
+		if err != nil {
+			logrus.Errorf("failed to lookup schemas for user %v: %v", user, err)
+			http.Error(rw, "schemas failed", http.StatusInternalServerError)
+			return
+		}
+
+		server.Handle(&types.APIRequest{
+			Request:  req,
+			Response: rw,
+			Schemas:  schemas,
+		})
+	})
+}
+
 func NewCollection(ctx context.Context, baseSchema *types.APISchemas, access accesscontrol.AccessSetLookup) *Collection {
 	return &Collection{
 		baseSchema: baseSchema,

pkg/schemaserver/parse/mux.go

@@ -7,16 +7,53 @@ import (
 	"github.com/rancher/steve/pkg/schemaserver/types"
 )
 
+type Vars struct {
+	Type      string
+	Name      string
+	Namespace string
+	Link      string
+	Prefix    string
+	Action    string
+}
+
+func Set(v Vars) mux.MatcherFunc {
+	return func(request *http.Request, match *mux.RouteMatch) bool {
+		if match.Vars == nil {
+			match.Vars = map[string]string{}
+		}
+		if v.Type != "" {
+			match.Vars["type"] = v.Type
+		}
+		if v.Name != "" {
+			match.Vars["name"] = v.Name
+		}
+		if v.Link != "" {
+			match.Vars["link"] = v.Link
+		}
+		if v.Prefix != "" {
+			match.Vars["prefix"] = v.Prefix
+		}
+		if v.Action != "" {
+			match.Vars["action"] = v.Action
+		}
+		if v.Namespace != "" {
+			match.Vars["namespace"] = v.Namespace
+		}
+		return true
+	}
+}
+
 func MuxURLParser(rw http.ResponseWriter, req *http.Request, schemas *types.APISchemas) (ParsedURL, error) {
 	vars := mux.Vars(req)
 	url := ParsedURL{
-		Type:   vars["type"],
+		Type:      vars["type"],
-		Name:   vars["name"],
+		Name:      vars["name"],
-		Link:   vars["link"],
+		Namespace: vars["namespace"],
-		Prefix: vars["prefix"],
+		Link:      vars["link"],
-		Method: req.Method,
+		Prefix:    vars["prefix"],
-		Action: vars["action"],
+		Method:    req.Method,
-		Query:  req.URL.Query(),
+		Action:    vars["action"],
+		Query:     req.URL.Query(),
 	}
 
 	return url, nil

pkg/schemaserver/parse/parse.go

@@ -24,6 +24,7 @@ var (
 type ParsedURL struct {
 	Type       string
 	Name       string
+	Namespace  string
 	Link       string
 	Method     string
 	Action     string
@@ -80,6 +81,9 @@ func Parse(apiOp *types.APIRequest, urlParser URLParser) error {
 	if apiOp.URLPrefix == "" {
 		apiOp.URLPrefix = parsedURL.Prefix
 	}
+	if apiOp.Namespace == "" {
+		apiOp.Namespace = parsedURL.Namespace
+	}
 
 	if apiOp.URLBuilder == nil {
 		// make error local to not override the outer error we have yet to check

pkg/schemaserver/server/server.go

@@ -6,9 +6,9 @@ import (
 	"github.com/rancher/steve/pkg/schemaserver/builtin"
 	"github.com/rancher/steve/pkg/schemaserver/handlers"
 	"github.com/rancher/steve/pkg/schemaserver/parse"
+	"github.com/rancher/steve/pkg/schemaserver/subscribe"
 	"github.com/rancher/steve/pkg/schemaserver/types"
 	"github.com/rancher/steve/pkg/schemaserver/writer"
-	"github.com/rancher/wrangler/pkg/merr"
 	"github.com/rancher/wrangler/pkg/schemas/validation"
 )
 
@@ -34,7 +34,6 @@ type Defaults struct {
 	CreateHandler types.RequestHandler
 	DeleteHandler types.RequestHandler
 	UpdateHandler types.RequestHandler
-	Store         types.Store
 	ErrorHandler  types.ErrorHandler
 }
 
@@ -70,6 +69,7 @@ func DefaultAPIServer() *Server {
 		URLParser: parse.MuxURLParser,
 	}
 
+	subscribe.Register(s.Schemas)
 	return s
 }
 
@@ -88,57 +88,6 @@ func (s *Server) setDefaults(ctx *types.APIRequest) {
 	}
 }
 
-func (s *Server) AddSchemas(schemas *types.APISchemas) error {
-	var errs []error
-
-	for _, schema := range schemas.Schemas {
-		if err := s.addSchema(*schema); err != nil {
-			errs = append(errs, err)
-		}
-	}
-
-	return merr.NewErrors(errs...)
-}
-
-func (s *Server) addSchema(schema types.APISchema) error {
-	s.setupDefaults(&schema)
-	return s.Schemas.AddSchema(schema)
-}
-
-func (s *Server) setupDefaults(schema *types.APISchema) {
-	if schema.Store == nil {
-		schema.Store = s.Defaults.Store
-	}
-
-	if schema.ListHandler == nil {
-		schema.ListHandler = s.Defaults.ListHandler
-	}
-
-	if schema.CreateHandler == nil {
-		schema.CreateHandler = s.Defaults.CreateHandler
-	}
-
-	if schema.ByIDHandler == nil {
-		schema.ByIDHandler = s.Defaults.ByIDHandler
-	}
-
-	if schema.UpdateHandler == nil {
-		schema.UpdateHandler = s.Defaults.UpdateHandler
-	}
-
-	if schema.DeleteHandler == nil {
-		schema.DeleteHandler = s.Defaults.DeleteHandler
-	}
-
-	if schema.ErrorHandler == nil {
-		schema.ErrorHandler = s.Defaults.ErrorHandler
-	}
-}
-
-func (s *Server) GetSchemas() *types.APISchemas {
-	return s.Schemas
-}
-
 func (s *Server) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 	s.Handle(&types.APIRequest{
 		Request:  req,

pkg/schemaserver/types/schemas.go

@@ -29,16 +29,30 @@ func (a *APISchemas) MustAddSchema(obj APISchema) *APISchemas {
 	return a
 }
 
-func (a *APISchemas) MustImportAndCustomize(obj interface{}, f func(*APISchema)) {
+func (a *APISchemas) addInternalSchema(schema *schemas.Schema) *APISchema {
-	schema, err := a.InternalSchemas.Import(obj)
-	if err != nil {
-		panic(err)
-	}
 	apiSchema := &APISchema{
 		Schema: schema,
 	}
 	a.Schemas[schema.ID] = apiSchema
 	a.addToIndex(apiSchema)
+
+	for _, f := range schema.ResourceFields {
+		if subType := a.InternalSchemas.Schema(f.Type); subType == nil {
+			continue
+		} else if _, ok := a.Schemas[subType.ID]; !ok {
+			a.addInternalSchema(subType)
+		}
+	}
+
+	return apiSchema
+}
+
+func (a *APISchemas) MustImportAndCustomize(obj interface{}, f func(*APISchema)) {
+	schema, err := a.InternalSchemas.Import(obj)
+	if err != nil {
+		panic(err)
+	}
+	apiSchema := a.addInternalSchema(schema)
 	f(apiSchema)
 }
 

pkg/server/config.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/rancher/steve/pkg/accesscontrol"
 	"github.com/rancher/steve/pkg/auth"
+	"github.com/rancher/steve/pkg/client"
 	"github.com/rancher/steve/pkg/schema"
 	"github.com/rancher/steve/pkg/schemaserver/types"
 	"github.com/rancher/steve/pkg/server/router"
@@ -29,6 +30,7 @@ type Server struct {
 
 	RestConfig *rest.Config
 
+	ClientFactory   *client.Factory
 	BaseSchemas     *types.APISchemas
 	AccessSetLookup accesscontrol.AccessSetLookup
 	SchemaTemplates []schema.Template

pkg/server/resources/schema.go

@@ -6,7 +6,6 @@ import (
 	"github.com/rancher/steve/pkg/clustercache"
 	"github.com/rancher/steve/pkg/schema"
 	"github.com/rancher/steve/pkg/schemaserver/store/apiroot"
-	"github.com/rancher/steve/pkg/schemaserver/subscribe"
 	"github.com/rancher/steve/pkg/schemaserver/types"
 	"github.com/rancher/steve/pkg/server/resources/apigroups"
 	"github.com/rancher/steve/pkg/server/resources/common"
@@ -16,7 +15,6 @@ import (
 
 func DefaultSchemas(baseSchema *types.APISchemas, discovery discovery.DiscoveryInterface, ccache clustercache.ClusterCache) *types.APISchemas {
 	counts.Register(baseSchema, ccache)
-	subscribe.Register(baseSchema)
 	apigroups.Register(baseSchema, discovery)
 	apiroot.Register(baseSchema, []string{"v1"}, []string{"proxy:/apis"})
 	return baseSchema

pkg/server/server.go

@@ -45,13 +45,18 @@ func setDefaults(server *Server) error {
 }
 
 func setup(ctx context.Context, server *Server) (http.Handler, *schema.Collection, error) {
-	if err := setDefaults(server); err != nil {
+	err := setDefaults(server)
+	if err != nil {
 		return nil, nil, err
 	}
 
-	cf, err := client.NewFactory(server.RestConfig, server.AuthMiddleware != nil)
+	cf := server.ClientFactory
-	if err != nil {
+	if cf == nil {
-		return nil, nil, err
+		cf, err = client.NewFactory(server.RestConfig, server.AuthMiddleware != nil)
+		if err != nil {
+			return nil, nil, err
+		}
+		server.ClientFactory = cf
 	}
 
 	asl := server.AccessSetLookup

pkg/server/store/proxy/proxy_store.go

@@ -32,8 +32,10 @@ var (
 type ClientGetter interface {
 	Client(ctx *types.APIRequest, schema *types.APISchema, namespace string) (dynamic.ResourceInterface, error)
 	AdminClient(ctx *types.APIRequest, schema *types.APISchema, namespace string) (dynamic.ResourceInterface, error)
-	ClientForWatch(ctx *types.APIRequest, schema *types.APISchema, namespace string) (dynamic.ResourceInterface, error)
+	TableClient(ctx *types.APIRequest, schema *types.APISchema, namespace string) (dynamic.ResourceInterface, error)
-	AdminClientForWatch(ctx *types.APIRequest, schema *types.APISchema, namespace string) (dynamic.ResourceInterface, error)
+	TableAdminClient(ctx *types.APIRequest, schema *types.APISchema, namespace string) (dynamic.ResourceInterface, error)
+	TableClientForWatch(ctx *types.APIRequest, schema *types.APISchema, namespace string) (dynamic.ResourceInterface, error)
+	TableAdminClientForWatch(ctx *types.APIRequest, schema *types.APISchema, namespace string) (dynamic.ResourceInterface, error)
 }
 
 type Store struct {
@@ -92,7 +94,7 @@ func toAPI(schema *types.APISchema, obj runtime.Object) types.APIObject {
 }
 
 func (s *Store) byID(apiOp *types.APIRequest, schema *types.APISchema, id string) (*unstructured.Unstructured, error) {
-	k8sClient, err := s.clientGetter.Client(apiOp, schema, apiOp.Namespace)
+	k8sClient, err := s.clientGetter.TableClient(apiOp, schema, apiOp.Namespace)
 	if err != nil {
 		return nil, err
 	}
@@ -184,7 +186,7 @@ func tableToObjects(obj map[string]interface{}) []unstructured.Unstructured {
 }
 
 func (s *Store) ByNames(apiOp *types.APIRequest, schema *types.APISchema, names sets.String) (types.APIObjectList, error) {
-	adminClient, err := s.clientGetter.AdminClient(apiOp, schema, apiOp.Namespace)
+	adminClient, err := s.clientGetter.TableAdminClient(apiOp, schema, apiOp.Namespace)
 	if err != nil {
 		return types.APIObjectList{}, err
 	}
@@ -206,7 +208,7 @@ func (s *Store) ByNames(apiOp *types.APIRequest, schema *types.APISchema, names
 }
 
 func (s *Store) List(apiOp *types.APIRequest, schema *types.APISchema) (types.APIObjectList, error) {
-	client, err := s.clientGetter.Client(apiOp, schema, apiOp.Namespace)
+	client, err := s.clientGetter.TableClient(apiOp, schema, apiOp.Namespace)
 	if err != nil {
 		return types.APIObjectList{}, err
 	}
@@ -287,7 +289,7 @@ func (s *Store) listAndWatch(apiOp *types.APIRequest, k8sClient dynamic.Resource
 }
 
 func (s *Store) WatchNames(apiOp *types.APIRequest, schema *types.APISchema, w types.WatchRequest, names sets.String) (chan types.APIEvent, error) {
-	adminClient, err := s.clientGetter.ClientForWatch(apiOp, schema, apiOp.Namespace)
+	adminClient, err := s.clientGetter.TableClientForWatch(apiOp, schema, apiOp.Namespace)
 	if err != nil {
 		return nil, err
 	}
@@ -310,7 +312,7 @@ func (s *Store) WatchNames(apiOp *types.APIRequest, schema *types.APISchema, w t
 }
 
 func (s *Store) Watch(apiOp *types.APIRequest, schema *types.APISchema, w types.WatchRequest) (chan types.APIEvent, error) {
-	client, err := s.clientGetter.ClientForWatch(apiOp, schema, apiOp.Namespace)
+	client, err := s.clientGetter.TableClientForWatch(apiOp, schema, apiOp.Namespace)
 	if err != nil {
 		return nil, err
 	}
@@ -374,7 +376,7 @@ func (s *Store) Create(apiOp *types.APIRequest, schema *types.APISchema, params
 	gvk := attributes.GVK(schema)
 	input["apiVersion"], input["kind"] = gvk.ToAPIVersionAndKind()
 
-	k8sClient, err := s.clientGetter.Client(apiOp, schema, ns)
+	k8sClient, err := s.clientGetter.TableClient(apiOp, schema, ns)
 	if err != nil {
 		return types.APIObject{}, err
 	}
@@ -395,7 +397,7 @@ func (s *Store) Update(apiOp *types.APIRequest, schema *types.APISchema, params
 	)
 
 	ns := types.Namespace(input)
-	k8sClient, err := s.clientGetter.Client(apiOp, schema, ns)
+	k8sClient, err := s.clientGetter.TableClient(apiOp, schema, ns)
 	if err != nil {
 		return types.APIObject{}, err
 	}
@@ -460,7 +462,7 @@ func (s *Store) Delete(apiOp *types.APIRequest, schema *types.APISchema, id stri
 		return types.APIObject{}, nil
 	}
 
-	k8sClient, err := s.clientGetter.Client(apiOp, schema, apiOp.Namespace)
+	k8sClient, err := s.clientGetter.TableClient(apiOp, schema, apiOp.Namespace)
 	if err != nil {
 		return types.APIObject{}, err
 	}

pkg/server/store/proxy/rbac_store.go

@@ -13,6 +13,19 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 )
 
+type filterKey struct{}
+
+func AddNamespaceConstraint(req *http.Request, names ...string) *http.Request {
+	set := sets.NewString(names...)
+	ctx := context.WithValue(req.Context(), filterKey{}, set)
+	return req.WithContext(ctx)
+}
+
+func getNamespaceConstraint(req *http.Request) (sets.String, bool) {
+	set, ok := req.Context().Value(filterKey{}).(sets.String)
+	return set, ok
+}
+
 type RBACStore struct {
 	*Store
 }
@@ -24,6 +37,34 @@ type Partition struct {
 }
 
 func isPassthrough(apiOp *types.APIRequest, schema *types.APISchema, verb string) ([]Partition, bool) {
+	partitions, passthrough := isPassthroughUnconstrained(apiOp, schema, verb)
+	namespaces, ok := getNamespaceConstraint(apiOp.Request)
+	if !ok {
+		return partitions, passthrough
+	}
+
+	var result []Partition
+
+	if passthrough {
+		for namespace := range namespaces {
+			result = append(result, Partition{
+				Namespace: namespace,
+				All:       true,
+			})
+		}
+		return result, false
+	}
+
+	for _, partition := range partitions {
+		if namespaces.Has(partition.Namespace) {
+			result = append(result, partition)
+		}
+	}
+
+	return result, false
+}
+
+func isPassthroughUnconstrained(apiOp *types.APIRequest, schema *types.APISchema, verb string) ([]Partition, bool) {
 	accessListByVerb, _ := attributes.Access(schema).(accesscontrol.AccessListByVerb)
 	if accessListByVerb.All(verb) {
 		return nil, true