fix: check write permission in legacy write path (#19980)

CHANGELOG.md

@@ -15,6 +15,7 @@
 1. [19972](https://github.com/influxdata/influxdb/pull/19972): Remove unused 'influx-command-path' option from upgrade command
 1. [19969](https://github.com/influxdata/influxdb/pull/19969): Check if CLI configs file already exists during upgrade
 1. [19967](https://github.com/influxdata/influxdb/pull/19967): Add 'log-level' option to upgrade command
+1. [19980](https://github.com/influxdata/influxdb/pull/19980): Fix authorization checks in the V1 /write API
 
 ## v2.0.0-rc.4 [2020-11-05]
 

cmd/influxd/launcher/launcher.go

@@ -954,8 +954,7 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 		}
 	}
 
-	// N.B. the BucketService used by the DBRP service doesn't perform authorization.
-	dbrpSvc := dbrp.NewAuthorizedService(dbrp.NewService(ctx, ts.BucketService, m.kvStore))
+	dbrpSvc := dbrp.NewAuthorizedService(dbrp.NewService(ctx, authorizer.NewBucketService(ts.BucketService), m.kvStore))
 
 	cm := iqlcontrol.NewControllerMetrics([]string{})
 	m.reg.MustRegister(cm.PrometheusCollectors()...)

http/legacy/write_handler.go

@@ -2,6 +2,7 @@ package legacy
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"net/http"
 
@@ -128,6 +129,11 @@ func (h *WriteHandler) handleWrite(w http.ResponseWriter, r *http.Request) {
 	}
 	span.LogKV("bucket_id", bucket.ID)
 
+	if err := checkBucketWritePermissions(auth, bucket.OrgID, bucket.ID); err != nil {
+		h.HandleHTTPError(ctx, err, sw)
+		return
+	}
+
 	parsed, err := points.NewParser(req.Precision).Parse(ctx, auth.OrgID, bucket.ID, req.Body)
 	if err != nil {
 		h.HandleHTTPError(ctx, err, sw)
@@ -158,6 +164,29 @@ func (h *WriteHandler) findBucket(ctx context.Context, orgID influxdb.ID, db, rp
 	return h.BucketService.FindBucketByID(ctx, mapping.BucketID)
 }
 
+// checkBucketWritePermissions checks an Authorizer for write permissions to a
+// specific Bucket.
+func checkBucketWritePermissions(auth influxdb.Authorizer, orgID, bucketID influxdb.ID) error {
+	p, err := influxdb.NewPermissionAtID(bucketID, influxdb.WriteAction, influxdb.BucketsResourceType, orgID)
+	if err != nil {
+		return &influxdb.Error{
+			Code: influxdb.EInternal,
+			Op:   opWriteHandler,
+			Msg:  fmt.Sprintf("unable to create permission for bucket: %v", err),
+			Err:  err,
+		}
+	}
+	if pset, err := auth.PermissionSet(); err != nil || !pset.Allowed(*p) {
+		return &influxdb.Error{
+			Code: influxdb.EForbidden,
+			Op:   opWriteHandler,
+			Msg:  "insufficient permissions for write",
+			Err:  err,
+		}
+	}
+	return nil
+}
+
 // findMapping finds a DBRPMappingV2 for the database and retention policy
 // combination.
 func (h *WriteHandler) findMapping(ctx context.Context, orgID influxdb.ID, db, rp string) (*influxdb.DBRPMappingV2, error) {

http/legacy/write_handler_test.go

@@ -13,7 +13,6 @@ import (
 
 	"github.com/golang/mock/gomock"
 	"github.com/influxdata/influxdb/v2"
-	"github.com/influxdata/influxdb/v2/authorizer"
 	pcontext "github.com/influxdata/influxdb/v2/context"
 	"github.com/influxdata/influxdb/v2/dbrp"
 	"github.com/influxdata/influxdb/v2/http/mocks"
@@ -82,7 +81,7 @@ func TestWriteHandler_BucketAndMappingExists(t *testing.T) {
 		recordWriteEvent,
 	)
 
-	perms := newPermissions(influxdb.BucketsResourceType, &orgID, nil)
+	perms := newPermissions(influxdb.WriteAction, influxdb.BucketsResourceType, &orgID, nil)
 	auth := newAuthorization(orgID, perms...)
 	ctx := pcontext.SetAuthorizer(context.Background(), auth)
 	r := newWriteRequest(ctx, lineProtocolBody)
@@ -94,7 +93,7 @@ func TestWriteHandler_BucketAndMappingExists(t *testing.T) {
 	handler := NewWriterHandler(&PointsWriterBackend{
 		HTTPErrorHandler:   DefaultErrorHandler,
 		Logger:             zaptest.NewLogger(t),
-		BucketService:      authorizer.NewBucketService(bucketService),
+		BucketService:      bucketService,
 		DBRPMappingService: dbrp.NewAuthorizedService(dbrpMappingSvc),
 		PointsWriter:       pointsWriter,
 		EventRecorder:      eventRecorder,
@@ -105,6 +104,79 @@ func TestWriteHandler_BucketAndMappingExists(t *testing.T) {
 	assert.Equal(t, "", w.Body.String())
 }
 
+func TestWriteHandler_BucketAndMappingExistsNoPermissions(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	var (
+		// Mocked Services
+		eventRecorder  = mocks.NewMockEventRecorder(ctrl)
+		dbrpMappingSvc = mocks.NewMockDBRPMappingServiceV2(ctrl)
+		bucketService  = mocks.NewMockBucketService(ctrl)
+		pointsWriter   = mocks.NewMockPointsWriter(ctrl)
+
+		// Found Resources
+		orgID  = generator.ID()
+		bucket = &influxdb.Bucket{
+			ID:                  generator.ID(),
+			OrgID:               orgID,
+			Name:                "mydb/autogen",
+			RetentionPolicyName: "autogen",
+			RetentionPeriod:     72 * time.Hour,
+		}
+		mapping = &influxdb.DBRPMappingV2{
+			OrganizationID:  orgID,
+			BucketID:        bucket.ID,
+			Database:        "mydb",
+			RetentionPolicy: "autogen",
+		}
+
+		lineProtocolBody = "m,t1=v1 f1=2 100"
+	)
+
+	findAutogenMapping := dbrpMappingSvc.
+		EXPECT().
+		FindMany(gomock.Any(), influxdb.DBRPMappingFilterV2{
+			OrgID:    &mapping.OrganizationID,
+			Database: &mapping.Database,
+		}).Return([]*influxdb.DBRPMappingV2{mapping}, 1, nil)
+
+	findBucketByID := bucketService.
+		EXPECT().
+		FindBucketByID(gomock.Any(), bucket.ID).Return(bucket, nil)
+
+	recordWriteEvent := eventRecorder.EXPECT().
+		Record(gomock.Any(), gomock.Any())
+
+	gomock.InOrder(
+		findAutogenMapping,
+		findBucketByID,
+		recordWriteEvent,
+	)
+
+	perms := newPermissions(influxdb.ReadAction, influxdb.BucketsResourceType, &orgID, nil)
+	auth := newAuthorization(orgID, perms...)
+	ctx := pcontext.SetAuthorizer(context.Background(), auth)
+	r := newWriteRequest(ctx, lineProtocolBody)
+	params := r.URL.Query()
+	params.Set("db", "mydb")
+	params.Set("rp", "")
+	r.URL.RawQuery = params.Encode()
+
+	handler := NewWriterHandler(&PointsWriterBackend{
+		HTTPErrorHandler:   DefaultErrorHandler,
+		Logger:             zaptest.NewLogger(t),
+		BucketService:      bucketService,
+		DBRPMappingService: dbrp.NewAuthorizedService(dbrpMappingSvc),
+		PointsWriter:       pointsWriter,
+		EventRecorder:      eventRecorder,
+	})
+	w := httptest.NewRecorder()
+	handler.ServeHTTP(w, r)
+	assert.Equal(t, http.StatusForbidden, w.Code)
+	assert.Equal(t, "{\"code\":\"forbidden\",\"message\":\"insufficient permissions for write\"}", w.Body.String())
+}
+
 func TestWriteHandler_MappingNotExists(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -150,7 +222,7 @@ func TestWriteHandler_MappingNotExists(t *testing.T) {
 		recordWriteEvent,
 	)
 
-	perms := newPermissions(influxdb.BucketsResourceType, &orgID, nil)
+	perms := newPermissions(influxdb.WriteAction, influxdb.BucketsResourceType, &orgID, nil)
 	auth := newAuthorization(orgID, perms...)
 	ctx := pcontext.SetAuthorizer(context.Background(), auth)
 	r := newWriteRequest(ctx, lineProtocolBody)
@@ -162,7 +234,7 @@ func TestWriteHandler_MappingNotExists(t *testing.T) {
 	handler := NewWriterHandler(&PointsWriterBackend{
 		HTTPErrorHandler:   DefaultErrorHandler,
 		Logger:             zaptest.NewLogger(t),
-		BucketService:      authorizer.NewBucketService(bucketService),
+		BucketService:      bucketService,
 		DBRPMappingService: dbrp.NewAuthorizedService(dbrpMappingSvc),
 		PointsWriter:       pointsWriter,
 		EventRecorder:      eventRecorder,
@@ -230,18 +302,10 @@ func (m pointsMatcher) String() string {
 	return fmt.Sprintf("%#v", m.points)
 }
 
-func newPermissions(resourceType influxdb.ResourceType, orgID, id *influxdb.ID) []influxdb.Permission {
+func newPermissions(action influxdb.Action, resourceType influxdb.ResourceType, orgID, id *influxdb.ID) []influxdb.Permission {
 	return []influxdb.Permission{
 		{
-			Action: influxdb.WriteAction,
-			Resource: influxdb.Resource{
-				Type:  resourceType,
-				OrgID: orgID,
-				ID:    id,
-			},
-		},
-		{
-			Action: influxdb.ReadAction,
+			Action: action,
 			Resource: influxdb.Resource{
 				Type:  resourceType,
 				OrgID: orgID,