Repository: osbuild/osbuild-composer
Pull request #1450 (2021-06-07 13:45:56)

X rh identity header filter

teg: Does this change the worker api?
Gundersanne: @ondrejbudai I made a unit test which mimics the oauth sso, proxy, and then also the worker server. Sadly I couldn't get it to work with the default fsjobqueue so I made a tiny mock queue.

When I try it with the regular queue it gets stuck in reflect.Select when trying to dequeue a job uuid, and I'm not sure why :/
teg: > @ondrejbudai I made a unit test which mimics the oauth sso, proxy, and then also the worker server.

Nice!

> Sadly I couldn't get it to work with the default fsjobqueue so I made a tiny mock queue.
> 
> When I try it with the regular queue it gets stuck in reflect.Select when trying to dequeue a job uuid, and I'm not sure why :/

Hm, this is a bit disconcerting, no? I have no idea either, but probably something we need to figure out (in the unlikely event there is an actual regression here somewhere)?

Patch:
diff --git a/cmd/osbuild-composer/composer.go b/cmd/osbuild-composer/composer.go
index eafc7c5a5..ec8e7e8b6 100644
--- a/cmd/osbuild-composer/composer.go
+++ b/cmd/osbuild-composer/composer.go
@@ -68,7 +68,7 @@ func NewComposer(config *ComposerConfigFile, stateDir, cacheDir string, logger *
 		return nil, fmt.Errorf("cannot create jobqueue: %v", err)
 	}
 
-	c.workers = worker.NewServer(c.logger, jobs, artifactsDir)
+	c.workers = worker.NewServer(c.logger, jobs, artifactsDir, c.config.WorkerAPI.IdentityFilter)
 
 	return &c, nil
 }
@@ -135,17 +135,21 @@ func (c *Composer) InitLocalWorker(l net.Listener) {
 }
 
 func (c *Composer) InitRemoteWorkers(cert, key string, l net.Listener) error {
-	tlsConfig, err := createTLSConfig(&connectionConfig{
-		CACertFile:     c.config.Worker.CA,
-		ServerKeyFile:  key,
-		ServerCertFile: cert,
-		AllowedDomains: c.config.Worker.AllowedDomains,
-	})
-	if err != nil {
-		return fmt.Errorf("Error creating TLS configuration for remote worker API: %v", err)
-	}
+	if len(c.config.WorkerAPI.IdentityFilter) > 0 {
+		c.workerListener = l
+	} else {
+		tlsConfig, err := createTLSConfig(&connectionConfig{
+			CACertFile:     c.config.Worker.CA,
+			ServerKeyFile:  key,
+			ServerCertFile: cert,
+			AllowedDomains: c.config.Worker.AllowedDomains,
+		})
+		if err != nil {
+			return fmt.Errorf("Error creating TLS configuration for remote worker API: %v", err)
+		}
 
-	c.workerListener = tls.NewListener(l, tlsConfig)
+		c.workerListener = tls.NewListener(l, tlsConfig)
+	}
 
 	return nil
 }
diff --git a/cmd/osbuild-composer/config.go b/cmd/osbuild-composer/config.go
index f77af1384..6a2f0c9f5 100644
--- a/cmd/osbuild-composer/config.go
+++ b/cmd/osbuild-composer/config.go
@@ -18,6 +18,9 @@ type ComposerConfigFile struct {
 	ComposerAPI struct {
 		IdentityFilter []string `toml:"identity_filter"`
 	} `toml:"composer_api"`
+	WorkerAPI struct {
+		IdentityFilter []string `toml:"identity_filter"`
+	} `toml:"worker_api"`
 }
 
 func LoadConfig(name string) (*ComposerConfigFile, error) {
diff --git a/cmd/osbuild-worker/main.go b/cmd/osbuild-worker/main.go
index 574b2aa56..61a15ff60 100644
--- a/cmd/osbuild-worker/main.go
+++ b/cmd/osbuild-worker/main.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/BurntSushi/toml"
 
+	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/upload/azure"
 	"github.com/osbuild/osbuild-composer/internal/upload/koji"
 	"github.com/osbuild/osbuild-composer/internal/worker"
@@ -90,12 +91,16 @@ func main() {
 		Azure *struct {
 			Credentials string `toml:"credentials"`
 		} `toml:"azure"`
+		Authentication *struct {
+			OAuthURL         string `toml:"oauth_url"`
+			OfflineTokenPath string `toml:"offline_token"`
+		} `toml:"authentication"`
 	}
 	var unix bool
 	flag.BoolVar(&unix, "unix", false, "Interpret 'address' as a path to a unix domain socket instead of a network address")
 
 	flag.Usage = func() {
-		fmt.Fprintf(flag.CommandLine.Output(), "Usage: %s [-unix] address\n", os.Args[0])
+		fmt.Fprintf(flag.CommandLine.Output(), "Usage: %s [-unix] address basepath\n", os.Args[0])
 		flag.PrintDefaults()
 		os.Exit(0)
 	}
@@ -142,6 +147,21 @@ func main() {
 	var client *worker.Client
 	if unix {
 		client = worker.NewClientUnix(address)
+	} else if config.Authentication != nil && config.Authentication.OfflineTokenPath != "" {
+		t, err := ioutil.ReadFile(config.Authentication.OfflineTokenPath)
+		if err != nil {
+			log.Fatalf("Could not read offline token: %v", err)
+		}
+		token := string(t)
+
+		if config.Authentication.OAuthURL == "" {
+			log.Fatal("OAuth URL should be specified together with the offline token")
+		}
+
+		client, err = worker.NewClient("https://"+address, nil, &token, &config.Authentication.OAuthURL)
+		if err != nil {
+			log.Fatalf("Error creating worker client: %v", err)
+		}
 	} else {
 		conf, err := createTLSConfig(&connectionConfig{
 			CACertFile:     "/etc/osbuild-composer/ca-crt.pem",
@@ -152,7 +172,7 @@ func main() {
 			log.Fatalf("Error creating TLS config: %v", err)
 		}
 
-		client, err = worker.NewClient("https://"+address, conf)
+		client, err = worker.NewClient("https://"+address, conf, nil, nil)
 		if err != nil {
 			log.Fatalf("Error creating worker client: %v", err)
 		}
@@ -209,7 +229,7 @@ func main() {
 
 	for {
 		fmt.Println("Waiting for a new job...")
-		job, err := client.RequestJob(acceptedJobTypes)
+		job, err := client.RequestJob(acceptedJobTypes, common.CurrentArch())
 		if err != nil {
 			log.Fatal(err)
 		}
diff --git a/docs/news/unreleased/worker-oauth2-support.md b/docs/news/unreleased/worker-oauth2-support.md
new file mode 100644
index 000000000..776b2be5a
--- /dev/null
+++ b/docs/news/unreleased/worker-oauth2-support.md
@@ -0,0 +1,5 @@
+# Workers: oauth2 support
+
+This change is mainly targeted for getting composer into `cloud.redhat.com`. It
+allows remote workers to connect to composer starting from a refresh token, and
+is offered as an alternative to the client certificate authentication.
diff --git a/internal/cloudapi/server.go b/internal/cloudapi/server.go
index 0850c1347..fb0da8e65 100644
--- a/internal/cloudapi/server.go
+++ b/internal/cloudapi/server.go
@@ -34,16 +34,6 @@ type Server struct {
 
 type contextKey int
 
-const (
-	identityHeaderKey contextKey = iota
-)
-
-type IdentityHeader struct {
-	Identity struct {
-		AccountNumber string `json:"account_number"`
-	} `json:"identity"`
-}
-
 // NewServer creates a new cloud server
 func NewServer(workers *worker.Server, rpmMetadata rpmmd.RPMMD, distros *distroregistry.Registry) *Server {
 	server := &Server{
@@ -72,6 +62,13 @@ func (server *Server) Handler(path string, identityFilter []string) http.Handler
 
 func (server *Server) VerifyIdentityHeader(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		const identityHeaderKey contextKey = iota
+		type identityHeader struct {
+			Identity struct {
+				AccountNumber string `json:"account_number"`
+			} `json:"identity"`
+		}
+
 		idHeaderB64 := r.Header["X-Rh-Identity"]
 		if len(idHeaderB64) != 1 {
 			http.Error(w, "Auth header is not present", http.StatusNotFound)
@@ -84,7 +81,7 @@ func (server *Server) VerifyIdentityHeader(next http.Handler) http.Handler {
 			return
 		}
 
-		var idHeader IdentityHeader
+		var idHeader identityHeader
 		err = json.Unmarshal([]byte(strings.TrimSuffix(fmt.Sprintf("%s", b64Result), "\n")), &idHeader)
 		if err != nil {
 			http.Error(w, "Auth header has incorrect format", http.StatusNotFound)
diff --git a/internal/mocks/rpmmd/fixtures.go b/internal/mocks/rpmmd/fixtures.go
index 9f1fc2bc9..f2bbea684 100644
--- a/internal/mocks/rpmmd/fixtures.go
+++ b/internal/mocks/rpmmd/fixtures.go
@@ -57,7 +57,7 @@ func createBaseWorkersFixture(tmpdir string) *worker.Server {
 	if err != nil {
 		panic(err)
 	}
-	return worker.NewServer(nil, q, "")
+	return worker.NewServer(nil, q, "", []string{})
 }
 
 func createBaseDepsolveFixture() []rpmmd.PackageSpec {
diff --git a/internal/worker/api/api.go b/internal/worker/api/api.go
index d379f8f2a..72bf9a544 100644
--- a/internal/worker/api/api.go
+++ b/internal/worker/api/api.go
@@ -3,3 +3,4 @@
 package api
 
 const BasePath = "/api/worker/v1"
+const CloudBasePath = "/api/composer-worker/v1"
diff --git a/internal/worker/client.go b/internal/worker/client.go
index fcb0697c6..be1856a28 100644
--- a/internal/worker/client.go
+++ b/internal/worker/client.go
@@ -10,15 +10,29 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"strings"
+	"sync"
+	"time"
 
 	"github.com/google/uuid"
-	"github.com/osbuild/osbuild-composer/internal/common"
+
 	"github.com/osbuild/osbuild-composer/internal/worker/api"
 )
 
+type bearerToken struct {
+	AccessToken     string `json:"access_token"`
+	ValidForSeconds int    `json:"expires_in"`
+}
+
 type Client struct {
-	server    *url.URL
-	requester *http.Client
+	server           *url.URL
+	requester        *http.Client
+	offlineToken     *string
+	oAuthURL         *string
+	lastTokenRefresh *time.Time
+	bearerToken      *bearerToken
+
+	tokenMu *sync.Mutex
 }
 
 type Job interface {
@@ -33,7 +47,7 @@ type Job interface {
 }
 
 type job struct {
-	requester        *http.Client
+	client           *Client
 	id               uuid.UUID
 	location         string
 	artifactLocation string
@@ -42,24 +56,33 @@ type job struct {
 	dynamicArgs      []json.RawMessage
 }
 
-func NewClient(baseURL string, conf *tls.Config) (*Client, error) {
+func NewClient(baseURL string, conf *tls.Config, offlineToken, oAuthURL *string) (*Client, error) {
 	server, err := url.Parse(baseURL)
 	if err != nil {
 		return nil, err
 	}
 
-	server, err = server.Parse(api.BasePath + "/")
+	bp := api.BasePath
+	if offlineToken != nil {
+		bp = api.CloudBasePath
+	}
+	server, err = server.Parse(bp + "/")
 	if err != nil {
 		panic(err)
 	}
 
-	requester := &http.Client{
-		Transport: &http.Transport{
+	if conf != nil && offlineToken != nil {
+		return nil, fmt.Errorf("error creating client, both tls and oauth are enabled")
+	}
+
+	requester := &http.Client{}
+	if conf != nil {
+		requester.Transport = &http.Transport{
 			TLSClientConfig: conf,
-		},
+		}
 	}
 
-	return &Client{server, requester}, nil
+	return &Client{server, requester, offlineToken, oAuthURL, nil, nil, &sync.Mutex{}}, nil
 }
 
 func NewClientUnix(path string) *Client {
@@ -81,10 +104,65 @@ func NewClientUnix(path string) *Client {
 		},
 	}
 
-	return &Client{server, requester}
+	return &Client{server, requester, nil, nil, nil, nil, nil}
+}
+
+// Note: Only call this function with Client.tokenMu locked!
+func (c *Client) refreshBearerToken() error {
+	if c.offlineToken == nil || c.oAuthURL == nil {
+		return fmt.Errorf("No offline token or oauth url available")
+	}
+
+	data := url.Values{}
+	data.Set("grant_type", "refresh_token")
+	data.Set("client_id", "rhsm-api")
+	data.Set("refresh_token", *c.offlineToken)
+
+	t := time.Now()
+	resp, err := http.Post(*c.oAuthURL, "application/x-www-form-urlencoded", strings.NewReader(data.Encode()))
+	if err != nil {
+		return err
+	}
+
+	var bt bearerToken
+	err = json.NewDecoder(resp.Body).Decode(&bt)
+	if err != nil {
+		return err
+	}
+
+	c.bearerToken = &bt
+	c.lastTokenRefresh = &t
+	return nil
+}
+
+func (c *Client) NewRequest(method, url string, body io.Reader) (*http.Request, error) {
+	req, err := http.NewRequest(method, url, body)
+	if err != nil {
+		return nil, err
+	}
+
+	// If we're using OAUTH, add the Bearer token
+	if c.offlineToken != nil {
+		// make sure we have a valid token
+		var d time.Duration
+		c.tokenMu.Lock()
+		defer c.tokenMu.Unlock()
+		if c.lastTokenRefresh != nil {
+			d = time.Since(*c.lastTokenRefresh)
+		}
+		if c.bearerToken == nil || d.Seconds() >= (float64(c.bearerToken.ValidForSeconds)*0.8) {
+			err = c.refreshBearerToken()
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", c.bearerToken.AccessToken))
+	}
+	return req, nil
 }
 
-func (c *Client) RequestJob(types []string) (Job, error) {
+func (c *Client) RequestJob(types []string, arch string) (Job, error) {
 	url, err := c.server.Parse("jobs")
 	if err != nil {
 		// This only happens when "jobs" cannot be parsed.
@@ -94,13 +172,19 @@ func (c *Client) RequestJob(types []string) (Job, error) {
 	var buf bytes.Buffer
 	err = json.NewEncoder(&buf).Encode(api.RequestJobJSONRequestBody{
 		Types: types,
-		Arch:  common.CurrentArch(),
+		Arch:  arch,
 	})
 	if err != nil {
 		panic(err)
 	}
 
-	response, err := c.requester.Post(url.String(), "application/json", &buf)
+	req, err := c.NewRequest("POST", url.String(), &buf)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Add("Content-Type", "application/json")
+
+	response, err := c.requester.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("error requesting job: %v", err)
 	}
@@ -127,7 +211,7 @@ func (c *Client) RequestJob(types []string) (Job, error) {
 	}
 
 	return &job{
-		requester:        c.requester,
+		client:           c,
 		id:               jr.Id,
 		jobType:          jr.Type,
 		args:             jr.Args,
@@ -174,14 +258,14 @@ func (j *job) Update(result interface{}) error {
 		panic(err)
 	}
 
-	req, err := http.NewRequest("PATCH", j.location, &buf)
+	req, err := j.client.NewRequest("PATCH", j.location, &buf)
 	if err != nil {
 		panic(err)
 	}
 
 	req.Header.Add("Content-Type", "application/json")
 
-	response, err := j.requester.Do(req)
+	response, err := j.client.requester.Do(req)
 	if err != nil {
 		return fmt.Errorf("error fetching job info: %v", err)
 	}
@@ -195,7 +279,12 @@ func (j *job) Update(result interface{}) error {
 }
 
 func (j *job) Canceled() (bool, error) {
-	response, err := j.requester.Get(j.location)
+	req, err := j.client.NewRequest("GET", j.location, nil)
+	if err != nil {
+		return false, err
+	}
+
+	response, err := j.client.requester.Do(req)
 	if err != nil {
 		return false, fmt.Errorf("error fetching job info: %v", err)
 	}
@@ -229,14 +318,14 @@ func (j *job) UploadArtifact(name string, reader io.Reader) error {
 		panic(err)
 	}
 
-	req, err := http.NewRequest("PUT", loc.String(), reader)
+	req, err := j.client.NewRequest("PUT", loc.String(), reader)
 	if err != nil {
 		return fmt.Errorf("cannot create request: %v", err)
 	}
 
 	req.Header.Add("Content-Type", "application/octet-stream")
 
-	response, err := j.requester.Do(req)
+	response, err := j.client.requester.Do(req)
 	if err != nil {
 		return fmt.Errorf("error uploading artifact: %v", err)
 	}
diff --git a/internal/worker/server.go b/internal/worker/server.go
index 727de708a..43f8a6a00 100644
--- a/internal/worker/server.go
+++ b/internal/worker/server.go
@@ -2,6 +2,7 @@ package worker
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -11,6 +12,7 @@ import (
 	"net/http"
 	"os"
 	"path"
+	"strings"
 	"sync"
 	"time"
 
@@ -22,9 +24,10 @@ import (
 )
 
 type Server struct {
-	jobs         jobqueue.JobQueue
-	logger       *log.Logger
-	artifactsDir string
+	jobs           jobqueue.JobQueue
+	logger         *log.Logger
+	artifactsDir   string
+	identityFilter []string
 
 	// Currently running jobs. Workers are not handed job ids, but
 	// independent tokens which serve as an indirection. This enables
@@ -48,12 +51,14 @@ type JobStatus struct {
 
 var ErrTokenNotExist = errors.New("worker token does not exist")
 
-func NewServer(logger *log.Logger, jobs jobqueue.JobQueue, artifactsDir string) *Server {
+func NewServer(logger *log.Logger, jobs jobqueue.JobQueue, artifactsDir string, identityFilter []string) *Server {
+
 	return &Server{
-		jobs:         jobs,
-		logger:       logger,
-		artifactsDir: artifactsDir,
-		running:      make(map[uuid.UUID]uuid.UUID),
+		jobs:           jobs,
+		logger:         logger,
+		artifactsDir:   artifactsDir,
+		identityFilter: identityFilter,
+		running:        make(map[uuid.UUID]uuid.UUID),
 	}
 }
 
@@ -68,14 +73,57 @@ func (s *Server) Handler() http.Handler {
 		e.DefaultHTTPErrorHandler(err, c)
 	}
 
+	var mws []echo.MiddlewareFunc
+	if len(s.identityFilter) > 0 {
+		mws = append(mws, s.VerifyIdentityHeader)
+	}
+
 	handler := apiHandlers{
 		server: s,
 	}
-	api.RegisterHandlers(e.Group(api.BasePath), &handler)
+	api.RegisterHandlers(e.Group(api.BasePath, mws...), &handler)
+	api.RegisterHandlers(e.Group(api.CloudBasePath, mws...), &handler)
 
 	return e
 }
 
+func (s *Server) VerifyIdentityHeader(nextHandler echo.HandlerFunc) echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		type identityHeader struct {
+			Identity struct {
+				AccountNumber string `json:"account_number"`
+			} `json:"identity"`
+		}
+
+		request := ctx.Request()
+
+		idHeaderB64 := request.Header["X-Rh-Identity"]
+		if len(idHeaderB64) != 1 {
+			return echo.NewHTTPError(http.StatusNotFound, "Auth header is not present")
+		}
+
+		b64Result, err := base64.StdEncoding.DecodeString(idHeaderB64[0])
+		if err != nil {
+			return echo.NewHTTPError(http.StatusNotFound, "Auth header has incorrect format")
+		}
+
+		var idHeader identityHeader
+		err = json.Unmarshal([]byte(strings.TrimSuffix(fmt.Sprintf("%s", b64Result), "\n")), &idHeader)
+		if err != nil {
+			return echo.NewHTTPError(http.StatusNotFound, "Auth header has incorrect format")
+		}
+
+		for _, i := range s.identityFilter {
+			if idHeader.Identity.AccountNumber == i {
+				ctx.Set("IdentityHeader", idHeader)
+				return nextHandler(ctx)
+
+			}
+		}
+		return echo.NewHTTPError(http.StatusNotFound, "Account not allowed")
+	}
+}
+
 func (s *Server) EnqueueOSBuild(arch string, job *OSBuildJob) (uuid.UUID, error) {
 	return s.jobs.Enqueue("osbuild:"+arch, job, nil)
 }
@@ -300,10 +348,15 @@ func (h *apiHandlers) RequestJob(ctx echo.Context) error {
 		return err
 	}
 
+	basePath := api.BasePath
+	if strings.HasPrefix(ctx.Path(), api.CloudBasePath) {
+		basePath = api.CloudBasePath
+	}
+
 	return ctx.JSON(http.StatusCreated, requestJobResponse{
 		Id:               jobId,
-		Location:         fmt.Sprintf("%s/jobs/%v", api.BasePath, token),
-		ArtifactLocation: fmt.Sprintf("%s/jobs/%v/artifacts/", api.BasePath, token),
+		Location:         fmt.Sprintf("%s/jobs/%v", basePath, token),
+		ArtifactLocation: fmt.Sprintf("%s/jobs/%v/artifacts/", basePath, token),
 		Type:             jobType,
 		Args:             jobArgs,
 		DynamicArgs:      dynamicJobArgs,


Test patch:
diff --git a/internal/test/helpers.go b/internal/test/helpers.go
index a194e33f1..e4e9e948c 100644
--- a/internal/test/helpers.go
+++ b/internal/test/helpers.go
@@ -50,9 +50,14 @@ func externalRequest(method, path, body string) *http.Response {
 	return resp
 }
 
-func internalRequest(api http.Handler, method, path, body string) *http.Response {
+func internalRequest(api http.Handler, method, path, body string, header map[string]string) *http.Response {
 	req := httptest.NewRequest(method, path, bytes.NewReader([]byte(body)))
 	req.Header.Set("Content-Type", "application/json")
+
+	for k, h := range header {
+		req.Header.Set(k, h)
+	}
+
 	resp := httptest.NewRecorder()
 	api.ServeHTTP(resp, req)
 
@@ -66,10 +71,14 @@ func SendHTTP(api http.Handler, external bool, method, path, body string) *http.
 		}
 		return externalRequest(method, path, body)
 	} else {
-		return internalRequest(api, method, path, body)
+		return internalRequest(api, method, path, body, map[string]string{})
 	}
 }
 
+func SendHTTPWithHeader(api http.Handler, method, path, body string, header map[string]string) *http.Response {
+	return internalRequest(api, method, path, body, header)
+}
+
 // this function serves to drop fields that shouldn't be tested from the unmarshalled json objects
 func dropFields(obj interface{}, fields ...string) {
 	switch v := obj.(type) {
diff --git a/internal/worker/server_test.go b/internal/worker/server_test.go
index 6b23a2e26..a7609362c 100644
--- a/internal/worker/server_test.go
+++ b/internal/worker/server_test.go
@@ -2,13 +2,17 @@ package worker_test
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"net/http/httptest"
 	"os"
+	"strings"
 	"testing"
 
 	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/osbuild/osbuild-composer/internal/distro"
@@ -18,12 +22,12 @@ import (
 	"github.com/osbuild/osbuild-composer/internal/worker"
 )
 
-func newTestServer(t *testing.T, tempdir string) *worker.Server {
+func newTestServer(t *testing.T, tempdir string, identities []string) *worker.Server {
 	q, err := fsjobqueue.New(tempdir)
 	if err != nil {
 		t.Fatalf("error creating fsjobqueue: %v", err)
 	}
-	return worker.NewServer(nil, q, "")
+	return worker.NewServer(nil, q, "", identities)
 }
 
 // Ensure that the status request returns OK.
@@ -32,7 +36,7 @@ func TestStatus(t *testing.T) {
 	require.NoError(t, err)
 	defer os.RemoveAll(tempdir)
 
-	server := newTestServer(t, tempdir)
+	server := newTestServer(t, tempdir, []string{})
 	handler := server.Handler()
 	test.TestRoute(t, handler, false, "GET", "/api/worker/v1/status", ``, http.StatusOK, `{"status":"OK"}`, "message")
 }
@@ -63,7 +67,7 @@ func TestErrors(t *testing.T) {
 	defer os.RemoveAll(tempdir)
 
 	for _, c := range cases {
-		server := newTestServer(t, tempdir)
+		server := newTestServer(t, tempdir, []string{})
 		handler := server.Handler()
 		test.TestRoute(t, handler, false, c.Method, c.Path, c.Body, c.ExpectedStatus, "{}", "message")
 	}
@@ -87,7 +91,7 @@ func TestCreate(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, tempdir)
+	server := newTestServer(t, tempdir, []string{})
 	handler := server.Handler()
 
 	_, err = server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: manifest})
@@ -116,7 +120,7 @@ func TestCancel(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, tempdir)
+	server := newTestServer(t, tempdir, []string{})
 	handler := server.Handler()
 
 	jobId, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: manifest})
@@ -157,7 +161,7 @@ func TestUpdate(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, tempdir)
+	server := newTestServer(t, tempdir, []string{})
 	handler := server.Handler()
 
 	jobId, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: manifest})
@@ -186,7 +190,7 @@ func TestArgs(t *testing.T) {
 	tempdir, err := ioutil.TempDir("", "worker-tests-")
 	require.NoError(t, err)
 	defer os.RemoveAll(tempdir)
-	server := newTestServer(t, tempdir)
+	server := newTestServer(t, tempdir, []string{})
 
 	job := worker.OSBuildJob{
 		Manifest:  manifest,
@@ -226,7 +230,7 @@ func TestUpload(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, tempdir)
+	server := newTestServer(t, tempdir, []string{})
 	handler := server.Handler()
 
 	jobID, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: manifest})
@@ -241,3 +245,112 @@ func TestUpload(t *testing.T) {
 
 	test.TestRoute(t, handler, false, "PUT", fmt.Sprintf("/api/worker/v1/jobs/%s/artifacts/foobar", token), `this is my artifact`, http.StatusOK, `?`)
 }
+
+func TestIdentities(t *testing.T) {
+	tempdir, err := ioutil.TempDir("", "worker-tests-")
+	require.NoError(t, err)
+	defer os.RemoveAll(tempdir)
+
+	// distroStruct := test_distro.New()
+	// arch, err := distroStruct.GetArch(test_distro.TestArchName)
+	// require.NoError(t, err)
+	// imageType, err := arch.GetImageType(test_distro.TestImageTypeName)
+	// require.NoError(t, err)
+	// manifest, err := imageType.Manifest(nil, distro.ImageOptions{Size: imageType.Size(0)}, nil, nil, 0)
+	// require.NoError(t, err)
+
+	server := newTestServer(t, tempdir, []string{"000000"})
+	handler := server.Handler()
+
+	// _, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: manifest})
+	// require.NoError(t, err)
+
+	test.TestRoute(t, handler, false, "GET", "/api/worker/v1/status", ``, http.StatusNotFound, `{"message":"Auth header is not present"}`, "message")
+
+	header := map[string]string{
+		"x-rh-identity": "eyJlbnRpdGxlbWVudHMiOnsiaW5zaWdodHMiOnsiaXNfZW50aXRsZWQiOnRydWV9LCJzbWFydF9tYW5hZ2VtZW50Ijp7ImlzX2VudGl0bGVkIjp0cnVlfSwib3BlbnNoaWZ0Ijp7ImlzX2VudGl0bGVkIjp0cnVlfSwiaHlicmlkIjp7ImlzX2VudGl0bGVkIjp0cnVlfSwibWlncmF0aW9ucyI6eyJpc19lbnRpdGxlZCI6dHJ1ZX0sImFuc2libGUiOnsiaXNfZW50aXRsZWQiOnRydWV9fSwiaWRlbnRpdHkiOnsiYWNjb3VudF9udW1iZXIiOiIwMDAwMDMiLCJ0eXBlIjoiVXNlciIsInVzZXIiOnsidXNlcm5hbWUiOiJ1c2VyIiwiZW1haWwiOiJ1c2VyQHVzZXIudXNlciIsImZpcnN0X25hbWUiOiJ1c2VyIiwibGFzdF9uYW1lIjoidXNlciIsImlzX2FjdGl2ZSI6dHJ1ZSwiaXNfb3JnX2FkbWluIjp0cnVlLCJpc19pbnRlcm5hbCI6dHJ1ZSwibG9jYWxlIjoiZW4tVVMifSwiaW50ZXJuYWwiOnsib3JnX2lkIjoiMDAwMDAwIn19fQo=",
+	}
+	response := test.SendHTTPWithHeader(handler, "GET", "/api/worker/v1/status", ``, header)
+	assert.Equal(t, 404, response.StatusCode, "status mismatch")
+
+	header = map[string]string{
+		"x-rh-identity": "eyJlbnRpdGxlbWVudHMiOnsiaW5zaWdodHMiOnsiaXNfZW50aXRsZWQiOnRydWV9LCJzbWFydF9tYW5hZ2VtZW50Ijp7ImlzX2VudGl0bGVkIjp0cnVlfSwib3BlbnNoaWZ0Ijp7ImlzX2VudGl0bGVkIjp0cnVlfSwiaHlicmlkIjp7ImlzX2VudGl0bGVkIjp0cnVlfSwibWlncmF0aW9ucyI6eyJpc19lbnRpdGxlZCI6dHJ1ZX0sImFuc2libGUiOnsiaXNfZW50aXRsZWQiOnRydWV9fSwiaWRlbnRpdHkiOnsiYWNjb3VudF9udW1iZXIiOiIwMDAwMDAiLCJ0eXBlIjoiVXNlciIsInVzZXIiOnsidXNlcm5hbWUiOiJ1c2VyIiwiZW1haWwiOiJ1c2VyQHVzZXIudXNlciIsImZpcnN0X25hbWUiOiJ1c2VyIiwibGFzdF9uYW1lIjoidXNlciIsImlzX2FjdGl2ZSI6dHJ1ZSwiaXNfb3JnX2FkbWluIjp0cnVlLCJpc19pbnRlcm5hbCI6dHJ1ZSwibG9jYWxlIjoiZW4tVVMifSwiaW50ZXJuYWwiOnsib3JnX2lkIjoiMDAwMDAwIn19fQ==",
+	}
+
+	response = test.SendHTTPWithHeader(handler, "GET", "/api/worker/v1/status", ``, header)
+	assert.Equal(t, 200, response.StatusCode, "status mismatch")
+}
+
+func TestOAuth(t *testing.T) {
+	tempdir, err := ioutil.TempDir("", "worker-tests-")
+	require.NoError(t, err)
+	defer os.RemoveAll(tempdir)
+
+	q, err := fsjobqueue.New(tempdir)
+	require.NoError(t, err)
+	workerServer := worker.NewServer(nil, q, tempdir, []string{"000000"})
+	handler := workerServer.Handler()
+
+	workSrv := httptest.NewServer(handler)
+	defer workSrv.Close()
+
+	/* Start a server which will act as a proxy, adding a valid identity header  */
+	proxySrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		require.Equal(t, "Bearer accessToken!", r.Header.Get("Authorization"))
+		r.Header.Set("x-rh-identity", "eyJlbnRpdGxlbWVudHMiOnsiaW5zaWdodHMiOnsiaXNfZW50aXRsZWQiOnRydWV9LCJzbWFydF9tYW5hZ2VtZW50Ijp7ImlzX2VudGl0bGVkIjp0cnVlfSwib3BlbnNoaWZ0Ijp7ImlzX2VudGl0bGVkIjp0cnVlfSwiaHlicmlkIjp7ImlzX2VudGl0bGVkIjp0cnVlfSwibWlncmF0aW9ucyI6eyJpc19lbnRpdGxlZCI6dHJ1ZX0sImFuc2libGUiOnsiaXNfZW50aXRsZWQiOnRydWV9fSwiaWRlbnRpdHkiOnsiYWNjb3VudF9udW1iZXIiOiIwMDAwMDAiLCJ0eXBlIjoiVXNlciIsInVzZXIiOnsidXNlcm5hbWUiOiJ1c2VyIiwiZW1haWwiOiJ1c2VyQHVzZXIudXNlciIsImZpcnN0X25hbWUiOiJ1c2VyIiwibGFzdF9uYW1lIjoidXNlciIsImlzX2FjdGl2ZSI6dHJ1ZSwiaXNfb3JnX2FkbWluIjp0cnVlLCJpc19pbnRlcm5hbCI6dHJ1ZSwibG9jYWxlIjoiZW4tVVMifSwiaW50ZXJuYWwiOnsib3JnX2lkIjoiMDAwMDAwIn19fQ==")
+		handler.ServeHTTP(w, r)
+	}))
+	defer proxySrv.Close()
+
+	offlineToken := "someOfflineToken"
+	/* Start oauth srv supplying the bearer token */
+	oauthSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		require.Equal(t, "POST", r.Method)
+		err = r.ParseForm()
+		require.NoError(t, err)
+
+		require.Equal(t, "refresh_token", r.FormValue("grant_type"))
+		require.Equal(t, "rhsm-api", r.FormValue("client_id"))
+		require.Equal(t, offlineToken, r.FormValue("refresh_token"))
+
+		bt := struct {
+			AccessToken     string `json:"access_token"`
+			ValidForSeconds int    `json:"expires_in"`
+		}{
+			"accessToken!",
+			900,
+		}
+
+		w.WriteHeader(http.StatusOK)
+		w.Header().Set("Content-Type", "application/json")
+		err = json.NewEncoder(w).Encode(bt)
+		require.NoError(t, err)
+	}))
+	defer oauthSrv.Close()
+
+	distroStruct := test_distro.New()
+	arch, err := distroStruct.GetArch(test_distro.TestArchName)
+	if err != nil {
+		t.Fatalf("error getting arch from distro: %v", err)
+	}
+	imageType, err := arch.GetImageType(test_distro.TestImageTypeName)
+	if err != nil {
+		t.Fatalf("error getting image type from arch: %v", err)
+	}
+	manifest, err := imageType.Manifest(nil, distro.ImageOptions{Size: imageType.Size(0)}, nil, nil, 0)
+	if err != nil {
+		t.Fatalf("error creating osbuild manifest: %v", err)
+	}
+
+	_, err = workerServer.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: manifest})
+	require.NoError(t, err)
+
+	client, err := worker.NewClient(proxySrv.URL, nil, &offlineToken, &oauthSrv.URL)
+	require.NoError(t, err)
+	job, err := client.RequestJob([]string{"osbuild"}, arch.Name())
+	require.NoError(t, err)
+	r := strings.NewReader("artifact contents")
+	require.NoError(t, job.UploadArtifact("some-artifact", r))
+	c, err := job.Canceled()
+	require.False(t, c)
+}