add upstream auth check (#8)

* add upstream auth check

* fix read token cache issue

* add careHeaders to webservices

api/v1alpha1/webservice_types.go

@@ -17,6 +17,8 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"time"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -52,6 +54,14 @@ type UpstreamHttpAuthService struct {
 	// +kubebuilder:default=Authorization
 	// WriteTokenTo specifies which header should carry token to upstream service
 	WriteTokenTo string `json:"writeTokenTo"`
+
+	// CareHeaders specifies which headers from the upstream should be added to the downstream response. 
+	// +optional
+	CareHeaders []string `json:"careHeaders,omitempty"`
+
+	// +kubebuilder:default="200ms"
+	// Timeout specifies the duration to wait before timing out the request to the upstream authentication service.
+	Timeout time.Duration `json:"timeout"`
 }
 
 // WebServiceStatus defines the observed state of WebService

config/crd/bases/cerberus.snappcloud.io_webservices.yaml

@@ -48,18 +48,31 @@ spec:
                   address:
                     description: Address of the upstream authentication service
                     type: string
+                  careHeaders:
+                    description: CareHeaders specifies which headers from the upstream
+                      should be added to the downstream response.
+                    items:
+                      type: string
+                    type: array
                   readTokenFrom:
                     default: Authorization
                     description: ReadTokenFrom specifies which header contains the
                       upstream Auth token in the request
                     type: string
+                  timeout:
+                    default: 200ms
+                    description: Timeout specifies the duration to wait before timing
+                      out the request to the upstream authentication service.
+                    format: int64
+                    type: integer
                   writeTokenTo:
                     default: Authorization
                     description: WriteTokenTo specifies which header should carry
                       token to upstream service
                     type: string
                 required:
                 - readTokenFrom
+                - timeout
                 - writeTokenTo
                 type: object
             type: object

go.mod

@@ -18,6 +18,7 @@ require (
 )
 
 require (
+	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/cncf/xds/go v0.0.0-20230607035331-e9ce68804cb4 // indirect

go.sum

@@ -40,6 +40,8 @@ github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRF
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
+github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so=
+github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A=

pkg/auth/authenticator.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/asaskevich/govalidator"
 	"github.com/go-logr/logr"
 	cerberusv1alpha1 "github.com/snapp-incubator/Cerberus/api/v1alpha1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -19,7 +20,8 @@ import (
 // Authenticator can generate cache from Kubernetes API server
 // and it implements envoy.CheckRequest interface
 type Authenticator struct {
-	logger logr.Logger
+	logger     logr.Logger
+	httpClient *http.Client
 
 	accessCache   *AccessCache
 	servicesCache *ServicesCache
@@ -87,6 +89,22 @@ const (
 	// CerberusReasonWebserviceNotFound means that given webservice in
 	// the request context is not listed by Cerberus
 	CerberusReasonWebserviceNotFound CerberusReason = "webservice-notfound"
+
+	// CerberusReasonInvalidUpstreamAddress means that requested webservice
+	// has an invalid upstream address in it's manifest
+	CerberusReasonInvalidUpstreamAddress CerberusReason = "invalid-auth-upstream"
+
+	// CerberusReasonSourceAuthTokenEmpty means that requested webservice
+	// does not contain source upstream auth lookup header in it's manifest
+	CerberusReasonSourceAuthTokenEmpty CerberusReason = "upstream-source-identifier-empty"
+
+	// CerberusReasonTargetAuthTokenEmpty means that requested webservice
+	// does not contain a target upstream auth lookup header in it's manifest
+	CerberusReasonTargetAuthTokenEmpty CerberusReason = "upstream-target-identifier-empty"
+
+	// CerberusReasonUpstreamAuthFailed means that the request to the specified
+	// upstream failed due to an unidentified issue
+	CerberusReasonUpstreamAuthFailed CerberusReason = "upstream-auth-failed"
 )
 
 //+kubebuilder:rbac:groups=cerberus.snappcloud.io,resources=accesstokens,verbs=get;list;watch;
@@ -211,25 +229,18 @@ func (a *Authenticator) UpdateCache(c client.Client, ctx context.Context, readOn
 
 // TestAccess will check if given AccessToken (identified by raw token in the request)
 // has access to given Webservice (identified by it's name) and returns proper CerberusReason
-func (a *Authenticator) TestAccess(wsvc string, token string) (bool, CerberusReason, ExtraHeaders) {
+func (a *Authenticator) TestAccess(wsvc ServicesCacheEntry, token string) (bool, CerberusReason, ExtraHeaders) {
 	a.cacheLock.RLock()
 	cacheReaders.Inc()
 	defer a.cacheLock.RUnlock()
 	defer cacheReaders.Dec()
 
 	newExtraHeaders := make(ExtraHeaders)
 
-	if wsvc == "" {
-		return false, CerberusReasonLookupEmpty, newExtraHeaders
-	}
 	if token == "" {
 		return false, CerberusReasonTokenEmpty, newExtraHeaders
 	}
 
-	if _, ok := (*a.servicesCache)[wsvc]; !ok {
-		return false, CerberusReasonWebserviceNotFound, newExtraHeaders
-	}
-
 	ac, ok := (*a.accessCache)[token]
 
 	if !ok {
@@ -238,7 +249,7 @@ func (a *Authenticator) TestAccess(wsvc string, token string) (bool, CerberusRea
 
 	newExtraHeaders["X-Cerberus-AccessToken"] = ac.AccessToken.ObjectMeta.Name
 
-	if _, ok := (*a.accessCache)[token].allowedServices[wsvc]; !ok {
+	if _, ok := (*a.accessCache)[token].allowedServices[wsvc.Name]; !ok {
 		return false, CerberusReasonUnauthorized, newExtraHeaders
 	}
 
@@ -247,31 +258,50 @@ func (a *Authenticator) TestAccess(wsvc string, token string) (bool, CerberusRea
 
 // readToken reads token from given Request object and
 // will return error if it not exists at expected header
-// BUG: TODO: accuire lock before accessing the cache
-func (a *Authenticator) readToken(request *Request) (bool, CerberusReason, string) {
-	wsvc := request.Context["webservice"]
-	res, ok := (*a.servicesCache)[wsvc]
-	if !ok {
-		return false, CerberusReasonWebserviceNotFound, ""
-	}
-	if res.Spec.LookupHeader == "" {
+func (a *Authenticator) readToken(request *Request, wsvc ServicesCacheEntry) (bool, CerberusReason, string) {
+	if wsvc.Spec.LookupHeader == "" {
 		return false, CerberusReasonLookupIdentifierEmpty, ""
 	}
-	token := request.Request.Header.Get(res.Spec.LookupHeader)
+	token := request.Request.Header.Get(wsvc.Spec.LookupHeader)
 	return true, "", token
 }
 
+// readService reads requested webservice from cache and
+// will return error if the object would not be found in cache
+func (a *Authenticator) readService(wsvc string) (bool, CerberusReason, ServicesCacheEntry) {
+	a.cacheLock.RLock()
+	cacheReaders.Inc()
+	defer a.cacheLock.RUnlock()
+	defer cacheReaders.Dec()
+
+	if wsvc == "" {
+		return false, CerberusReasonLookupEmpty, ServicesCacheEntry{}
+	}
+
+	res, ok := (*a.servicesCache)[wsvc]
+	if !ok {
+		return false, CerberusReasonWebserviceNotFound, ServicesCacheEntry{}
+	}
+	return true, "", res
+}
+
 // Check is the function which is used to Authenticate and Respond to gRPC envoy.CheckRequest
 func (a *Authenticator) Check(ctx context.Context, request *Request) (*Response, error) {
 	reqStartTime := time.Now()
 	wsvc := request.Context["webservice"]
-
-	ok, reason, token := a.readToken(request)
 	var extraHeaders ExtraHeaders
 	var httpStatusCode int
+	var token string
 
+	ok, reason, wsvcCacheEntry := a.readService(wsvc)
 	if ok {
-		ok, reason, extraHeaders = a.TestAccess(wsvc, token)
+		ok, reason, token = a.readToken(request, wsvcCacheEntry)
+		if ok {
+			ok, reason, extraHeaders = a.TestAccess(wsvcCacheEntry, token)
+			if ok && hasUpstreamAuth(wsvcCacheEntry) {
+				ok, reason = a.checkServiceUpstreamAuth(wsvcCacheEntry, request, &extraHeaders)
+			}
+		}
 	}
 
 	// TODO: remove this line
@@ -308,7 +338,8 @@ func (a *Authenticator) Check(ctx context.Context, request *Request) (*Response,
 // currently it's not returning any error
 func NewAuthenticator(logger logr.Logger) (*Authenticator, error) {
 	a := Authenticator{
-		logger: logger,
+		logger:     logger,
+		httpClient: &http.Client{},
 	}
 	return &a, nil
 }
@@ -348,3 +379,66 @@ func CheckDomain(domain string, domainAllowedList []string) (bool, error) {
 	}
 	return false, nil
 }
+
+// checkServiceUpstreamAuth function is designed to validate the request through
+// the upstream authentication for a given webservice
+func (a *Authenticator) checkServiceUpstreamAuth(service ServicesCacheEntry, request *Request, extraHeaders *ExtraHeaders) (bool, CerberusReason) {
+	serviceUpstreamAuthCalls.Inc()
+
+	if service.Spec.UpstreamHttpAuth.ReadTokenFrom == "" {
+		return false, CerberusReasonSourceAuthTokenEmpty
+	}
+	if service.Spec.UpstreamHttpAuth.WriteTokenTo == "" {
+		return false, CerberusReasonTargetAuthTokenEmpty
+	}
+	if !govalidator.IsRequestURL(service.Spec.UpstreamHttpAuth.Address) {
+		return false, CerberusReasonInvalidUpstreamAddress
+	}
+
+	token := request.Request.Header.Get(service.Spec.UpstreamHttpAuth.ReadTokenFrom)
+
+	// TODO: get http method from webservice crd
+	req, err := http.NewRequest("GET", service.Spec.UpstreamHttpAuth.Address, nil)
+	if err != nil {
+		return false, CerberusReasonUpstreamAuthFailed
+	}
+
+	req.Header = http.Header{
+		service.Spec.UpstreamHttpAuth.WriteTokenTo: {token},
+		"Content-Type": {"application/json"},
+	}
+
+	a.httpClient.Timeout = service.Spec.UpstreamHttpAuth.Timeout
+	reqStart := time.Now()
+	resp, err := a.httpClient.Do(req)
+	reqDuration := time.Since(reqStart)
+	if err != nil {
+		return false, CerberusReasonUpstreamAuthFailed
+	}
+
+	labels := StatusLabel(resp.StatusCode)
+	upstreamAuthRequestDuration.With(labels).Observe(reqDuration.Seconds())
+
+	if resp.StatusCode != http.StatusOK {
+		return false, CerberusReasonUnauthorized
+	}
+	// add requested careHeaders to extraHeaders for response
+	for header, values := range resp.Header {
+		for _, careHeader := range service.Spec.UpstreamHttpAuth.CareHeaders {
+			if header == careHeader {
+				if len(values) > 0 {
+					(*extraHeaders)[header] = values[0]
+				}
+				break
+			}
+		}
+	}
+
+	return true, CerberusReasonOK
+}
+
+// hasUpstreamAuth evaluates whether the provided webservice
+// upstreamauth instance is considered empty or not
+func hasUpstreamAuth(service ServicesCacheEntry) bool {
+	return service.Spec.UpstreamHttpAuth.Address != ""
+}

pkg/auth/metrics.go

@@ -3,6 +3,7 @@ package auth
 import (
 	"github.com/prometheus/client_golang/prometheus"
 	"sigs.k8s.io/controller-runtime/pkg/metrics"
+	"strconv"
 )
 
 const (
@@ -14,6 +15,7 @@ const (
 	MetricsKindWebservice              = "webservice"
 	MetricsKindAccessToken             = "accesstoken"
 	MetricsKindWebserviceAccessBinding = "webserviceaccessbinding"
+	StatusCode                         = "status_code"
 
 	MetricsCheckRequestVersion2 = "v2"
 	MetricsCheckRequestVersion3 = "v3"
@@ -100,6 +102,21 @@ var (
 		},
 		[]string{ObjectKindLabel},
 	)
+
+	serviceUpstreamAuthCalls = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "upstream_auth_calls_total",
+			Help: "The total number of checkServiceUpstreamAuth function calls",
+		})
+
+	upstreamAuthRequestDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name:    "upstream_auth_request_duration_seconds",
+			Help:    "Duration of the UpstreamAuth Requests in seconds",
+			Buckets: DurationBuckets,
+		},
+		[]string{StatusCode},
+	)
 )
 
 func init() {
@@ -114,6 +131,8 @@ func init() {
 		accessCacheEntries,
 		webserviceCacheEntries,
 		fetchObjectListLatency,
+		serviceUpstreamAuthCalls,
+		upstreamAuthRequestDuration,
 	)
 }
 
@@ -128,3 +147,9 @@ func KindLabel(kind string) prometheus.Labels {
 	labels[ObjectKindLabel] = kind
 	return labels
 }
+
+func StatusLabel(status int) prometheus.Labels {
+	labels := prometheus.Labels{}
+	labels[StatusCode] = strconv.Itoa(status)
+	return labels
+}