api/{cloud,worker}: used channel name based on JWT claims for new jobs

This commit implements multi-tenancy. A tenant is defined based on a value
from JWT claims. The key of this value must be specified in the configuration
file. This allows us to pick different values when using multiple SSOs.

Let me explain more in depth how this works:

Cloud API gets a new compose request. Firstly, it extracts a tenant name from
JWT claims. The considered claims are configured as an array in
cloud_api.jwt.tenant_provider_fields in composer's config file. The channel
name for all jobs belonging to this compose is created by `"org-" + tenant`.

Why is the channel prefixed by "org-"? To give us options in the future. I can
imagine the request having a channel override. This basically means that
multiple tenants can share a channel. A real use-case for this is multiple
Fedora projects sharing one pool of workers.

Why this commit adds a whole new cloud_api section to the config? Because the
current config is a mess and we should stop adding new stuff into the koji
section. As the Koji API is basically deprecated, we will need to remove it
soon nevertheless.

Signed-off-by: Ondřej Budai <ondrej@budai.cz>

cmd/osbuild-composer/composer.go

@@ -55,7 +55,9 @@ func NewComposer(config *ComposerConfigFile, stateDir, cacheDir string) (*Compos
 	}
 
 	workerConfig := worker.Config{
-		BasePath: config.Worker.BasePath,
+		BasePath:             config.Worker.BasePath,
+		JWTEnabled:           config.Worker.EnableJWT,
+		TenantProviderFields: config.Worker.JWTTenantProviderFields,
 	}
 
 	var err error
@@ -125,7 +127,8 @@ func (c *Composer) InitWeldr(repoPaths []string, weldrListener net.Listener,
 func (c *Composer) InitAPI(cert, key string, enableTLS bool, enableMTLS bool, enableJWT bool, l net.Listener) error {
 	config := v2.ServerConfig{
 		AWSBucket:            c.config.Koji.AWS.Bucket,
-		TenantProviderFields: c.config.CloudAPI.JWT.TenantProviderFields,
+		JWTEnabled:           c.config.Koji.EnableJWT,
+		TenantProviderFields: c.config.Koji.JWTTenantProviderFields,
 	}
 
 	c.api = cloudapi.NewServer(c.workers, c.distros, config)

cmd/osbuild-composer/config.go

@@ -19,15 +19,16 @@ type ComposerConfigFile struct {
 }
 
 type KojiAPIConfig struct {
-	AllowedDomains []string  `toml:"allowed_domains"`
+	AllowedDomains          []string  `toml:"allowed_domains"`
-	CA             string    `toml:"ca"`
+	CA                      string    `toml:"ca"`
-	EnableTLS      bool      `toml:"enable_tls"`
+	EnableTLS               bool      `toml:"enable_tls"`
-	EnableMTLS     bool      `toml:"enable_mtls"`
+	EnableMTLS              bool      `toml:"enable_mtls"`
-	EnableJWT      bool      `toml:"enable_jwt"`
+	EnableJWT               bool      `toml:"enable_jwt"`
-	JWTKeysURLs    []string  `toml:"jwt_keys_urls"`
+	JWTKeysURLs             []string  `toml:"jwt_keys_urls"`
-	JWTKeysCA      string    `toml:"jwt_ca_file"`
+	JWTKeysCA               string    `toml:"jwt_ca_file"`
-	JWTACLFile     string    `toml:"jwt_acl_file"`
+	JWTACLFile              string    `toml:"jwt_acl_file"`
-	AWS            AWSConfig `toml:"aws_config"`
+	JWTTenantProviderFields []string  `toml:"jwt_tenant_provider_fields"`
+	AWS                     AWSConfig `toml:"aws_config"`
 }
 
 type AWSConfig struct {
@@ -35,24 +36,25 @@ type AWSConfig struct {
 }
 
 type WorkerAPIConfig struct {
-	AllowedDomains    []string `toml:"allowed_domains"`
+	AllowedDomains          []string `toml:"allowed_domains"`
-	CA                string   `toml:"ca"`
+	CA                      string   `toml:"ca"`
-	RequestJobTimeout string   `toml:"request_job_timeout"`
+	RequestJobTimeout       string   `toml:"request_job_timeout"`
-	BasePath          string   `toml:"base_path"`
+	BasePath                string   `toml:"base_path"`
-	EnableArtifacts   bool     `toml:"enable_artifacts"`
+	EnableArtifacts         bool     `toml:"enable_artifacts"`
-	PGHost            string   `toml:"pg_host" env:"PGHOST"`
+	PGHost                  string   `toml:"pg_host" env:"PGHOST"`
-	PGPort            string   `toml:"pg_port" env:"PGPORT"`
+	PGPort                  string   `toml:"pg_port" env:"PGPORT"`
-	PGDatabase        string   `toml:"pg_database" env:"PGDATABASE"`
+	PGDatabase              string   `toml:"pg_database" env:"PGDATABASE"`
-	PGUser            string   `toml:"pg_user" env:"PGUSER"`
+	PGUser                  string   `toml:"pg_user" env:"PGUSER"`
-	PGPassword        string   `toml:"pg_password" env:"PGPASSWORD"`
+	PGPassword              string   `toml:"pg_password" env:"PGPASSWORD"`
-	PGSSLMode         string   `toml:"pg_ssl_mode" env:"PGSSLMODE"`
+	PGSSLMode               string   `toml:"pg_ssl_mode" env:"PGSSLMODE"`
-	PGMaxConns        int      `toml:"pg_max_conns" env:"PGMAXCONNS"`
+	PGMaxConns              int      `toml:"pg_max_conns" env:"PGMAXCONNS"`
-	EnableTLS         bool     `toml:"enable_tls"`
+	EnableTLS               bool     `toml:"enable_tls"`
-	EnableMTLS        bool     `toml:"enable_mtls"`
+	EnableMTLS              bool     `toml:"enable_mtls"`
-	EnableJWT         bool     `toml:"enable_jwt"`
+	EnableJWT               bool     `toml:"enable_jwt"`
-	JWTKeysURLs       []string `toml:"jwt_keys_urls"`
+	JWTKeysURLs             []string `toml:"jwt_keys_urls"`
-	JWTKeysCA         string   `toml:"jwt_ca_file"`
+	JWTKeysCA               string   `toml:"jwt_ca_file"`
-	JWTACLFile        string   `toml:"jwt_acl_file"`
+	JWTACLFile              string   `toml:"jwt_acl_file"`
+	JWTTenantProviderFields []string `toml:"jwt_tenant_provider_fields"`
 }
 
 type WeldrAPIConfig struct {

internal/auth/jwt.go

@@ -0,0 +1,39 @@
+package auth
+
+import (
+	"context"
+	"errors"
+
+	"github.com/golang-jwt/jwt"
+	"github.com/openshift-online/ocm-sdk-go/authentication"
+)
+
+var NoJWTError = errors.New("request doesn't contain JWT")
+var NoKeyError = errors.New("cannot find key in jwt claims")
+
+// GetFromClaims returns a value of JWT claim with the specified key
+//
+// Caller can specify multiple keys. The value of first one that exists and is
+// non-empty is returned.
+//
+// If no claim is found, NoKeyError is returned
+func GetFromClaims(ctx context.Context, keys []string) (string, error) {
+	token, err := authentication.TokenFromContext(ctx)
+	if err != nil {
+		return "", err
+	} else if token == nil {
+		return "", NoJWTError
+	}
+
+	claims := token.Claims.(jwt.MapClaims)
+	for _, f := range keys {
+		value, exists := claims[f]
+		valueStr, isString := value.(string)
+		if exists && isString && valueStr != "" {
+			return valueStr, nil
+		}
+
+	}
+
+	return "", NoKeyError
+}

internal/auth/jwt_test.go

@@ -0,0 +1,66 @@
+package auth_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang-jwt/jwt"
+	"github.com/openshift-online/ocm-sdk-go/authentication"
+	"github.com/stretchr/testify/require"
+
+	"github.com/osbuild/osbuild-composer/internal/auth"
+)
+
+func TestChannelFromContext(t *testing.T) {
+	// tokens generated by https://jwt.io/ and signed by "osbuild"
+	tests := []struct {
+		name           string
+		token          string
+		value          string
+		expectedFields []string
+		err            error
+	}{
+		{
+			name:           "rh-org-id=42",
+			token:          "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyaC1vcmctaWQiOiI0MiJ9.D5EwgcrlCPcPamM9hL63bWI7xxr0YVWxsJ4f80toQv4",
+			value:          "42",
+			expectedFields: []string{"rh-org-id"},
+			err:            nil,
+		},
+		{
+			name:           "no rh-org-id",
+			token:          "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.e30.AmoXfoVMgoq4H-XsD7lTGgY6QJCW1914aYlmGnj7wtY",
+			value:          "",
+			expectedFields: []string{"rh-org-id"},
+			err:            auth.NoKeyError,
+		},
+		{
+			name:           "no rh-org-id but account_id=123",
+			token:          "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2NvdW50X2lkIjoiMTIzIn0.fng__koaJeF3Ef6E8kFOKCMm6U2MTwyFQ4s0G4LBUss",
+			value:          "123",
+			expectedFields: []string{"rh-org-id", "account_id"},
+			err:            nil,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			claims := jwt.MapClaims{}
+			token, err := jwt.ParseWithClaims(tt.token, claims, func(token *jwt.Token) (interface{}, error) {
+				return []byte("osbuild"), nil
+			})
+			require.NoError(t, err)
+
+			ctx := authentication.ContextWithToken(context.Background(), token)
+			channel, err := auth.GetFromClaims(ctx, tt.expectedFields)
+			require.Equal(t, tt.err, err)
+			require.Equal(t, tt.value, channel)
+
+		})
+	}
+
+	t.Run("no jwt token in context", func(t *testing.T) {
+		channel, err := auth.GetFromClaims(context.Background(), []string{"osbuild!"})
+		require.ErrorIs(t, err, auth.NoJWTError)
+		require.Equal(t, "", channel)
+	})
+}

internal/cloudapi/v2/errors.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/labstack/echo/v4"
+
 	"github.com/osbuild/osbuild-composer/internal/prometheus"
 )
 
@@ -40,6 +41,7 @@ const (
 	ErrorInvalidNumberOfImageBuilds   ServiceErrorCode = 25
 	ErrorInvalidJobType               ServiceErrorCode = 26
 	ErrorInvalidOSTreeParams          ServiceErrorCode = 27
+	ErrorTenantNotFound               ServiceErrorCode = 28
 
 	// Internal errors, these are bugs
 	ErrorFailedToInitializeBlueprint              ServiceErrorCode = 1000
@@ -104,6 +106,7 @@ func getServiceErrors() serviceErrors {
 		serviceError{ErrorInvalidJobType, http.StatusNotFound, "Requested job has invalid type"},
 		serviceError{ErrorInvalidNumberOfImageBuilds, http.StatusBadRequest, "Compose request has unsupported number of image builds"},
 		serviceError{ErrorInvalidOSTreeParams, http.StatusBadRequest, "Invalid OSTree parameters or parameter combination"},
+		serviceError{ErrorTenantNotFound, http.StatusBadRequest, "Tenant not found in JWT claims"},
 
 		serviceError{ErrorFailedToInitializeBlueprint, http.StatusInternalServerError, "Failed to initialize blueprint"},
 		serviceError{ErrorFailedToGenerateManifestSeed, http.StatusInternalServerError, "Failed to generate manifest seed"},

internal/cloudapi/v2/v2.go

@@ -18,6 +18,7 @@ import (
 	"github.com/labstack/echo/v4/middleware"
 	"github.com/sirupsen/logrus"
 
+	"github.com/osbuild/osbuild-composer/internal/auth"
 	"github.com/osbuild/osbuild-composer/internal/blueprint"
 	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/distro"
@@ -40,7 +41,9 @@ type Server struct {
 }
 
 type ServerConfig struct {
-	AWSBucket string
+	AWSBucket            string
+	TenantProviderFields []string
+	JWTEnabled           bool
 }
 
 type apiHandlers struct {
@@ -165,6 +168,18 @@ func (h *apiHandlers) PostCompose(ctx echo.Context) error {
 		return err
 	}
 
+	// channel is empty if JWT is not enabled
+	var channel string
+	if h.server.config.JWTEnabled {
+		tenant, err := auth.GetFromClaims(ctx.Request().Context(), h.server.config.TenantProviderFields)
+		if err != nil {
+			return HTTPErrorWithInternal(ErrorTenantNotFound, err)
+		}
+
+		// prefix the tenant to prevent collisions if support for specifying channels in a request is ever added
+		channel = "org-" + tenant
+	}
+
 	distribution := h.server.distros.GetDistro(request.Distribution)
 	if distribution == nil {
 		return HTTPError(ErrorUnsupportedDistribution)
@@ -469,12 +484,12 @@ func (h *apiHandlers) PostCompose(ctx echo.Context) error {
 
 	var id uuid.UUID
 	if request.Koji != nil {
-		id, err = enqueueKojiCompose(h.server.workers, uint64(request.Koji.TaskId), request.Koji.Server, request.Koji.Name, request.Koji.Version, request.Koji.Release, distribution, bp, manifestSeed, irs, "")
+		id, err = enqueueKojiCompose(h.server.workers, uint64(request.Koji.TaskId), request.Koji.Server, request.Koji.Name, request.Koji.Version, request.Koji.Release, distribution, bp, manifestSeed, irs, channel)
 		if err != nil {
 			return err
 		}
 	} else {
-		id, err = enqueueCompose(h.server.workers, distribution, bp, manifestSeed, irs, "")
+		id, err = enqueueCompose(h.server.workers, distribution, bp, manifestSeed, irs, channel)
 		if err != nil {
 			return err
 		}

internal/cloudapi/v2/v2_test.go

@@ -25,14 +25,15 @@ import (
 func newV2Server(t *testing.T, dir string) (*v2.Server, *worker.Server, context.CancelFunc) {
 	q, err := fsjobqueue.New(dir)
 	require.NoError(t, err)
-	workerServer := worker.NewServer(nil, q, worker.Config{BasePath: "/api/worker/v1"})
+	workerServer := worker.NewServer(nil, q, worker.Config{BasePath: "/api/worker/v1", TenantProviderFields: []string{"rh-org-id"}})
 
 	distros, err := distro_mock.NewDefaultRegistry()
 	require.NoError(t, err)
 	require.NotNil(t, distros)
 
 	config := v2.ServerConfig{
-		AWSBucket: "image-builder.service",
+		AWSBucket:            "image-builder.service",
+		TenantProviderFields: []string{"rh-org-id"},
 	}
 	v2Server := v2.NewServer(workerServer, distros, config)
 	require.NotNil(t, v2Server)

internal/worker/api/errors.go

@@ -22,6 +22,7 @@ const (
 	ErrorNotAcceptable        ServiceErrorCode = 13
 	ErrorErrorNotFound        ServiceErrorCode = 14
 	ErrorInvalidJobType       ServiceErrorCode = 15
+	ErrorTenantNotFound       ServiceErrorCode = 16
 	// ErrorTokenNotFound ServiceErrorCode = 6
 
 	// internal errors
@@ -74,6 +75,7 @@ func getServiceErrors() serviceErrors {
 		serviceError{ErrorNotAcceptable, http.StatusNotAcceptable, "Only 'application/json' content is supported"},
 		serviceError{ErrorErrorNotFound, http.StatusNotFound, "Error with given id not found"},
 		serviceError{ErrorInvalidJobType, http.StatusBadRequest, "Requested job type cannot be dequeued"},
+		serviceError{ErrorTenantNotFound, http.StatusBadRequest, "Tenant not found in JWT claims"},
 
 		serviceError{ErrorUnspecified, http.StatusInternalServerError, "Unspecified internal error "},
 		serviceError{ErrorNotHTTPError, http.StatusInternalServerError, "Error is not an instance of HTTPError"},

internal/worker/server.go

@@ -20,6 +20,7 @@ import (
 	"github.com/labstack/echo/v4/middleware"
 	"github.com/sirupsen/logrus"
 
+	"github.com/osbuild/osbuild-composer/internal/auth"
 	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/jobqueue"
 	"github.com/osbuild/osbuild-composer/internal/prometheus"
@@ -45,9 +46,11 @@ var ErrJobNotRunning = errors.New("job isn't running")
 var ErrInvalidJobType = errors.New("job has invalid type")
 
 type Config struct {
-	ArtifactsDir      string
+	ArtifactsDir         string
-	RequestJobTimeout time.Duration
+	RequestJobTimeout    time.Duration
-	BasePath          string
+	BasePath             string
+	JWTEnabled           bool
+	TenantProviderFields []string
 }
 
 func NewServer(logger *log.Logger, jobs jobqueue.JobQueue, config Config) *Server {
@@ -554,7 +557,19 @@ func (h *apiHandlers) RequestJob(ctx echo.Context) error {
 		return err
 	}
 
-	jobId, token, jobType, jobArgs, dynamicJobArgs, err := h.server.RequestJob(ctx.Request().Context(), body.Arch, body.Types, []string{""})
+	// channel is empty if JWT is not enabled
+	var channel string
+	if h.server.config.JWTEnabled {
+		tenant, err := auth.GetFromClaims(ctx.Request().Context(), h.server.config.TenantProviderFields)
+		if err != nil {
+			return api.HTTPErrorWithInternal(api.ErrorTenantNotFound, err)
+		}
+
+		// prefix the tenant to prevent collisions if support for specifying channels in a request is ever added
+		channel = "org-" + tenant
+	}
+
+	jobId, jobToken, jobType, jobArgs, dynamicJobArgs, err := h.server.RequestJob(ctx.Request().Context(), body.Arch, body.Types, []string{channel})
 	if err != nil {
 		if err == jobqueue.ErrDequeueTimeout {
 			return ctx.JSON(http.StatusNoContent, api.ObjectReference{
@@ -584,8 +599,8 @@ func (h *apiHandlers) RequestJob(ctx echo.Context) error {
 			Id:   jobId.String(),
 			Kind: "RequestJob",
 		},
-		Location:         fmt.Sprintf("%s/jobs/%v", api.BasePath, token),
+		Location:         fmt.Sprintf("%s/jobs/%v", api.BasePath, jobToken),
-		ArtifactLocation: fmt.Sprintf("%s/jobs/%v/artifacts/", api.BasePath, token),
+		ArtifactLocation: fmt.Sprintf("%s/jobs/%v/artifacts/", api.BasePath, jobToken),
 		Type:             jobType,
 		Args:             respArgs,
 		DynamicArgs:      respDynArgs,

test/cases/api.sh

@@ -1265,6 +1265,7 @@ enable_jwt = true
 jwt_keys_urls = ["https://localhost:8080/certs"]
 jwt_ca_file = "/etc/osbuild-composer/ca-crt.pem"
 jwt_acl_file = ""
+jwt_tenant_provider_fields = ["rh-org-id"]
 [worker]
 pg_host = "localhost"
 pg_port = "5432"
@@ -1278,6 +1279,7 @@ enable_mtls = false
 enable_jwt = true
 jwt_keys_urls = ["https://localhost:8080/certs"]
 jwt_ca_file = "/etc/osbuild-composer/ca-crt.pem"
+jwt_tenant_provider_fields = ["rh-org-id"]
 EOF
 
 cat <<EOF | sudo tee "/etc/osbuild-worker/token"