rpc,tenantcapabilities: allow cross tenant reads in shared service

pkg/kv/kvserver/tenantrate/limiter_test.go

@@ -667,7 +667,7 @@ func (ts *testState) BindReader(tenantcapabilities.Reader) {}
 
 var _ tenantcapabilities.Authorizer = &testState{}
 
-func (ts *testState) HasCrossTenantRead(tenID roachpb.TenantID) bool {
+func (ts *testState) HasCrossTenantRead(ctx context.Context, tenID roachpb.TenantID) bool {
 	return false
 }
 
@@ -789,7 +789,7 @@ type fakeAuthorizer struct{}
 
 var _ tenantcapabilities.Authorizer = &fakeAuthorizer{}
 
-func (fakeAuthorizer) HasCrossTenantRead(tenID roachpb.TenantID) bool {
+func (fakeAuthorizer) HasCrossTenantRead(ctx context.Context, tenID roachpb.TenantID) bool {
 	return false
 }
 

pkg/multitenant/tenantcapabilities/interfaces.go

@@ -42,7 +42,7 @@ type Reader interface {
 // usage pattern over a timespan.
 type Authorizer interface {
 	// HasCrossTenantRead returns true if a tenant can read other tenant spans.
-	HasCrossTenantRead(tenID roachpb.TenantID) bool
+	HasCrossTenantRead(ctx context.Context, tenID roachpb.TenantID) bool
 
 	// HasCapabilityForBatch returns an error if a tenant, referenced by its ID,
 	// is not allowed to execute the supplied batch request given the capabilities

pkg/multitenant/tenantcapabilities/tenantcapabilitiesauthorizer/allow_everything.go

@@ -30,7 +30,9 @@ func NewAllowEverythingAuthorizer() *AllowEverythingAuthorizer {
 }
 
 // HasCrossTenantRead returns true if a tenant can read from other tenants.
-func (n *AllowEverythingAuthorizer) HasCrossTenantRead(tenID roachpb.TenantID) bool {
+func (n *AllowEverythingAuthorizer) HasCrossTenantRead(
+	ctx context.Context, tenID roachpb.TenantID,
+) bool {
 	return true
 }
 

pkg/multitenant/tenantcapabilities/tenantcapabilitiesauthorizer/allow_nothing.go

@@ -31,7 +31,9 @@ func NewAllowNothingAuthorizer() *AllowNothingAuthorizer {
 }
 
 // HasCrossTenantRead returns true if a tenant can read from other tenants.
-func (n *AllowNothingAuthorizer) HasCrossTenantRead(tenID roachpb.TenantID) bool {
+func (n *AllowNothingAuthorizer) HasCrossTenantRead(
+	ctx context.Context, tenID roachpb.TenantID,
+) bool {
 	return false
 }
 

pkg/multitenant/tenantcapabilities/tenantcapabilitiesauthorizer/authorizer.go

@@ -98,8 +98,22 @@ func New(settings *cluster.Settings, knobs *tenantcapabilities.TestingKnobs) *Au
 }
 
 // HasCrossTenantRead returns true if a tenant can read from other tenants.
-func (a *Authorizer) HasCrossTenantRead(tenID roachpb.TenantID) bool {
-	return tenID.IsSystem()
+func (a *Authorizer) HasCrossTenantRead(ctx context.Context, tenID roachpb.TenantID) bool {
+	if tenID.IsSystem() {
+		// The system tenant has access to all request types.
+		return true
+	}
+	_, mode := a.getMode(ctx, tenID)
+	switch mode {
+	case authorizerModeOn, authorizerModeV222:
+		return false
+	case authorizerModeAllowAll:
+		return true
+	default:
+		err := errors.AssertionFailedf("unknown authorizer mode: %d", mode)
+		logcrash.ReportOrPanic(ctx, &a.settings.SV, "%v", err)
+		return false
+	}
 }
 
 // HasCapabilityForBatch implements the tenantcapabilities.Authorizer interface.

pkg/rpc/auth_tenant.go

@@ -64,7 +64,7 @@ func (a tenantAuthorizer) authorize(
 		return a.authBatch(ctx, sv, tenID, req.(*kvpb.BatchRequest))
 
 	case "/cockroach.roachpb.Internal/RangeLookup":
-		return a.authRangeLookup(tenID, req.(*kvpb.RangeLookupRequest))
+		return a.authRangeLookup(ctx, tenID, req.(*kvpb.RangeLookupRequest))
 
 	case "/cockroach.roachpb.Internal/RangeFeed", "/cockroach.roachpb.Internal/MuxRangeFeed":
 		return a.authRangeFeed(tenID, req.(*kvpb.RangeFeedRequest))
@@ -123,22 +123,22 @@ func (a tenantAuthorizer) authorize(
 		return a.authTenant(tenID)
 
 	case "/cockroach.server.serverpb.Status/SpanStats":
-		return a.authSpanStats(tenID, req.(*roachpb.SpanStatsRequest))
+		return a.authSpanStats(ctx, tenID, req.(*roachpb.SpanStatsRequest))
 
 	case "/cockroach.roachpb.Internal/GetSpanConfigs":
-		return a.authGetSpanConfigs(tenID, req.(*roachpb.GetSpanConfigsRequest))
+		return a.authGetSpanConfigs(ctx, tenID, req.(*roachpb.GetSpanConfigsRequest))
 
 	case "/cockroach.roachpb.Internal/SpanConfigConformance":
-		return a.authSpanConfigConformance(tenID, req.(*roachpb.SpanConfigConformanceRequest))
+		return a.authSpanConfigConformance(ctx, tenID, req.(*roachpb.SpanConfigConformanceRequest))
 
 	case "/cockroach.roachpb.Internal/GetAllSystemSpanConfigsThatApply":
 		return a.authGetAllSystemSpanConfigsThatApply(tenID, req.(*roachpb.GetAllSystemSpanConfigsThatApplyRequest))
 
 	case "/cockroach.roachpb.Internal/UpdateSpanConfigs":
-		return a.authUpdateSpanConfigs(tenID, req.(*roachpb.UpdateSpanConfigsRequest))
+		return a.authUpdateSpanConfigs(ctx, tenID, req.(*roachpb.UpdateSpanConfigsRequest))
 
 	case "/cockroach.roachpb.Internal/GetRangeDescriptors":
-		return a.authGetRangeDescriptors(tenID, req.(*kvpb.GetRangeDescriptorsRequest))
+		return a.authGetRangeDescriptors(ctx, tenID, req.(*kvpb.GetRangeDescriptorsRequest))
 
 	case "/cockroach.server.serverpb.Status/HotRangesV2":
 		return a.authHotRangesV2(tenID)
@@ -199,7 +199,7 @@ func (a tenantAuthorizer) authBatch(
 	tenSpan := tenantPrefix(tenID)
 
 	if outsideTenant(rSpan, tenSpan) {
-		if args.IsReadOnly() && a.capabilitiesAuthorizer.HasCrossTenantRead(tenID) {
+		if args.IsReadOnly() && a.capabilitiesAuthorizer.HasCrossTenantRead(ctx, tenID) {
 			return nil
 		}
 		return spanErr(rSpan, tenSpan)
@@ -208,16 +208,16 @@ func (a tenantAuthorizer) authBatch(
 }
 
 func (a tenantAuthorizer) authGetRangeDescriptors(
-	tenID roachpb.TenantID, args *kvpb.GetRangeDescriptorsRequest,
+	ctx context.Context, tenID roachpb.TenantID, args *kvpb.GetRangeDescriptorsRequest,
 ) error {
-	return validateSpan(tenID, args.Span, true, a)
+	return validateSpan(ctx, tenID, args.Span, true, a)
 }
 
 func (a tenantAuthorizer) authSpanStats(
-	tenID roachpb.TenantID, args *roachpb.SpanStatsRequest,
+	ctx context.Context, tenID roachpb.TenantID, args *roachpb.SpanStatsRequest,
 ) error {
 	for _, span := range args.Spans {
-		err := validateSpan(tenID, span, true, a)
+		err := validateSpan(ctx, tenID, span, true, a)
 		if err != nil {
 			return err
 		}
@@ -228,12 +228,12 @@ func (a tenantAuthorizer) authSpanStats(
 // authRangeLookup authorizes the provided tenant to invoke the RangeLookup RPC
 // with the provided args.
 func (a tenantAuthorizer) authRangeLookup(
-	tenID roachpb.TenantID, args *kvpb.RangeLookupRequest,
+	ctx context.Context, tenID roachpb.TenantID, args *kvpb.RangeLookupRequest,
 ) error {
 	tenSpan := tenantPrefix(tenID)
 	if !tenSpan.ContainsKey(args.Key) {
 		// Allow it anyway if the tenant can read other tenants.
-		if a.capabilitiesAuthorizer.HasCrossTenantRead(tenID) {
+		if a.capabilitiesAuthorizer.HasCrossTenantRead(ctx, tenID) {
 			return nil
 		}
 		return authErrorf("requested key %s not fully contained in tenant keyspace %s", args.Key, tenSpan)
@@ -351,10 +351,10 @@ func (a tenantAuthorizer) authGetAllSystemSpanConfigsThatApply(
 // authGetSpanConfigs authorizes the provided tenant to invoke the
 // GetSpanConfigs RPC with the provided args.
 func (a tenantAuthorizer) authGetSpanConfigs(
-	tenID roachpb.TenantID, args *roachpb.GetSpanConfigsRequest,
+	ctx context.Context, tenID roachpb.TenantID, args *roachpb.GetSpanConfigsRequest,
 ) error {
 	for _, target := range args.Targets {
-		if err := validateSpanConfigTarget(tenID, target, true, a); err != nil {
+		if err := validateSpanConfigTarget(ctx, tenID, target, true, a); err != nil {
 			return err
 		}
 	}
@@ -364,15 +364,15 @@ func (a tenantAuthorizer) authGetSpanConfigs(
 // authUpdateSpanConfigs authorizes the provided tenant to invoke the
 // UpdateSpanConfigs RPC with the provided args.
 func (a tenantAuthorizer) authUpdateSpanConfigs(
-	tenID roachpb.TenantID, args *roachpb.UpdateSpanConfigsRequest,
+	ctx context.Context, tenID roachpb.TenantID, args *roachpb.UpdateSpanConfigsRequest,
 ) error {
 	for _, entry := range args.ToUpsert {
-		if err := validateSpanConfigTarget(tenID, entry.Target, false, a); err != nil {
+		if err := validateSpanConfigTarget(ctx, tenID, entry.Target, false, a); err != nil {
 			return err
 		}
 	}
 	for _, target := range args.ToDelete {
-		if err := validateSpanConfigTarget(tenID, target, false, a); err != nil {
+		if err := validateSpanConfigTarget(ctx, tenID, target, false, a); err != nil {
 			return err
 		}
 	}
@@ -393,10 +393,10 @@ func (a tenantAuthorizer) authHotRangesV2(tenID roachpb.TenantID) error {
 // authSpanConfigConformance authorizes the provided tenant to invoke the
 // SpanConfigConformance RPC with the provided args.
 func (a tenantAuthorizer) authSpanConfigConformance(
-	tenID roachpb.TenantID, args *roachpb.SpanConfigConformanceRequest,
+	ctx context.Context, tenID roachpb.TenantID, args *roachpb.SpanConfigConformanceRequest,
 ) error {
 	for _, sp := range args.Spans {
-		if err := validateSpan(tenID, sp, false, a); err != nil {
+		if err := validateSpan(ctx, tenID, sp, false, a); err != nil {
 			return err
 		}
 	}
@@ -434,7 +434,11 @@ func (a tenantAuthorizer) authTSDBQuery(
 // wholly contained within the tenant keyspace and system span config targets
 // must be well-formed.
 func validateSpanConfigTarget(
-	tenID roachpb.TenantID, spanConfigTarget roachpb.SpanConfigTarget, read bool, a tenantAuthorizer,
+	ctx context.Context,
+	tenID roachpb.TenantID,
+	spanConfigTarget roachpb.SpanConfigTarget,
+	read bool,
+	a tenantAuthorizer,
 ) error {
 	validateSystemTarget := func(target roachpb.SystemSpanConfigTarget) error {
 		if target.SourceTenantID != tenID {
@@ -463,23 +467,25 @@ func validateSpanConfigTarget(
 
 	switch spanConfigTarget.Union.(type) {
 	case *roachpb.SpanConfigTarget_Span:
-		return validateSpan(tenID, *spanConfigTarget.GetSpan(), read, a)
+		return validateSpan(ctx, tenID, *spanConfigTarget.GetSpan(), read, a)
 	case *roachpb.SpanConfigTarget_SystemSpanConfigTarget:
 		return validateSystemTarget(*spanConfigTarget.GetSystemSpanConfigTarget())
 	default:
 		return errors.AssertionFailedf("unknown span config target type")
 	}
 }
 
-func validateSpan(tenID roachpb.TenantID, sp roachpb.Span, isRead bool, a tenantAuthorizer) error {
+func validateSpan(
+	ctx context.Context, tenID roachpb.TenantID, sp roachpb.Span, isRead bool, a tenantAuthorizer,
+) error {
 	tenSpan := tenantPrefix(tenID)
 	rSpan, err := keys.SpanAddr(sp)
 	if err != nil {
 		return authError(err.Error())
 	}
 	if outsideTenant(rSpan, tenSpan) {
 		// Allow it anyway if the tenant can read other tenants.
-		if isRead && a.capabilitiesAuthorizer.HasCrossTenantRead(tenID) {
+		if isRead && a.capabilitiesAuthorizer.HasCrossTenantRead(ctx, tenID) {
 			return nil
 		}
 		return spanErr(rSpan, tenSpan)

pkg/rpc/auth_test.go

@@ -1120,7 +1120,7 @@ func (m mockAuthorizer) HasProcessDebugCapability(
 	return errors.New("tenant does not have capability")
 }
 
-func (m mockAuthorizer) HasCrossTenantRead(tenID roachpb.TenantID) bool {
+func (m mockAuthorizer) HasCrossTenantRead(ctx context.Context, tenID roachpb.TenantID) bool {
 	return m.hasCrossTenantRead
 }