Upstash integration backend (#4597)

Co-authored-by: Stefan McShane <[email protected]>

api/server/handlers/oauth_callback/upstash.go

@@ -0,0 +1,198 @@
+package oauth_callback
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/porter-dev/porter/api/server/handlers"
+	"github.com/porter-dev/porter/api/server/shared"
+	"github.com/porter-dev/porter/api/server/shared/apierrors"
+	"github.com/porter-dev/porter/api/server/shared/config"
+	"github.com/porter-dev/porter/internal/models/integrations"
+	"github.com/porter-dev/porter/internal/telemetry"
+)
+
+// OAuthCallbackUpstashHandler is the handler responding to the upstash oauth callback
+type OAuthCallbackUpstashHandler struct {
+	handlers.PorterHandlerReadWriter
+}
+
+// UpstashApiKeyEndpoint is the endpoint to fetch the upstash developer api key
+// nolint:gosec // Not a security key
+const UpstashApiKeyEndpoint = "https://api.upstash.com/apikey"
+
+// NewOAuthCallbackUpstashHandler generates a new OAuthCallbackUpstashHandler
+func NewOAuthCallbackUpstashHandler(
+	config *config.Config,
+	decoderValidator shared.RequestDecoderValidator,
+	writer shared.ResultWriter,
+) *OAuthCallbackUpstashHandler {
+	return &OAuthCallbackUpstashHandler{
+		PorterHandlerReadWriter: handlers.NewDefaultPorterHandler(config, decoderValidator, writer),
+	}
+}
+
+// ServeHTTP gets the upstash oauth token from the callback code, uses it to create a developer api token, then creates a new upstash integration
+func (p *OAuthCallbackUpstashHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	ctx, span := telemetry.NewSpan(r.Context(), "serve-oauth-callback-upstash")
+	defer span.End()
+
+	r = r.Clone(ctx)
+
+	session, err := p.Config().Store.Get(r, p.Config().ServerConf.CookieName)
+	if err != nil {
+		err = telemetry.Error(ctx, span, err, "session could not be retrieved")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusInternalServerError))
+		return
+	}
+
+	if _, ok := session.Values["state"]; !ok {
+		err = telemetry.Error(ctx, span, nil, "state not found in session")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusInternalServerError))
+		return
+	}
+
+	if r.URL.Query().Get("state") != session.Values["state"] {
+		err = telemetry.Error(ctx, span, nil, "state does not match")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusInternalServerError))
+		return
+	}
+
+	projID, ok := session.Values["project_id"].(uint)
+	if !ok {
+		err = telemetry.Error(ctx, span, nil, "project id not found in session")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusInternalServerError))
+		return
+	}
+	telemetry.WithAttributes(span,
+		telemetry.AttributeKV{Key: "project-id", Value: projID},
+	)
+
+	if projID == 0 {
+		err = telemetry.Error(ctx, span, nil, "project id not found in session")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusInternalServerError))
+		return
+	}
+
+	code := r.URL.Query().Get("code")
+	if code == "" {
+		err = telemetry.Error(ctx, span, nil, "code not found in query params")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusForbidden))
+		return
+	}
+
+	token, err := p.Config().UpstashConf.Exchange(ctx, code)
+	if err != nil {
+		err = telemetry.Error(ctx, span, err, "exchange failed")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusForbidden))
+		return
+	}
+
+	if !token.Valid() {
+		err = telemetry.Error(ctx, span, nil, "invalid token")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusForbidden))
+		return
+	}
+
+	// make an http call to https://api.upstash.com/apikey with authorization: bearer <access_token>
+	// to get the api key
+	apiKey, err := fetchUpstashApiKey(ctx, token.AccessToken)
+	if err != nil {
+		err = telemetry.Error(ctx, span, err, "error fetching upstash api key")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusInternalServerError))
+		return
+	}
+
+	oauthInt := integrations.UpstashIntegration{
+		SharedOAuthModel: integrations.SharedOAuthModel{
+			AccessToken:  []byte(token.AccessToken),
+			RefreshToken: []byte(token.RefreshToken),
+			Expiry:       token.Expiry,
+		},
+		ProjectID:       projID,
+		DeveloperApiKey: []byte(apiKey),
+	}
+
+	_, err = p.Repo().UpstashIntegration().Insert(ctx, oauthInt)
+	if err != nil {
+		err = telemetry.Error(ctx, span, err, "error creating oauth integration")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusInternalServerError))
+		return
+	}
+
+	redirect := "/dashboard"
+	if redirectStr, ok := session.Values["redirect_uri"].(string); ok && redirectStr != "" {
+		redirectURI, err := url.Parse(redirectStr)
+		if err == nil {
+			redirect = fmt.Sprintf("%s?%s", redirectURI.Path, redirectURI.RawQuery)
+		}
+	}
+	http.Redirect(w, r, redirect, http.StatusFound)
+}
+
+// UpstashApiKeyRequest is the request body to fetch the upstash developer api key
+type UpstashApiKeyRequest struct {
+	Name string `json:"name"`
+}
+
+// UpstashApiKeyResponse is the response body to fetch the upstash developer api key
+type UpstashApiKeyResponse struct {
+	ApiKey string `json:"api_key"`
+}
+
+func fetchUpstashApiKey(ctx context.Context, accessToken string) (string, error) {
+	ctx, span := telemetry.NewSpan(ctx, "fetch-upstash-api-key")
+	defer span.End()
+
+	data := UpstashApiKeyRequest{
+		Name: fmt.Sprintf("PORTER_API_KEY_%d", time.Now().Unix()),
+	}
+
+	jsonData, err := json.Marshal(data)
+	if err != nil {
+		return "", telemetry.Error(ctx, span, err, "error marshalling request body")
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, UpstashApiKeyEndpoint, bytes.NewBuffer(jsonData))
+	if err != nil {
+		return "", telemetry.Error(ctx, span, err, "error creating request")
+	}
+
+	// Set the Authorization header
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", accessToken))
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return "", telemetry.Error(ctx, span, err, "error sending request")
+	}
+	defer resp.Body.Close() // nolint: errcheck
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "status-code", Value: resp.StatusCode})
+	if resp.StatusCode != http.StatusOK {
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "read-response-body-error", Value: err.Error()})
+		}
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "response-body", Value: string(body)})
+		return "", telemetry.Error(ctx, span, nil, "unexpected status code")
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", telemetry.Error(ctx, span, err, "error reading response body")
+	}
+
+	var responseData UpstashApiKeyResponse
+	err = json.Unmarshal(body, &responseData)
+	if err != nil {
+		return "", telemetry.Error(ctx, span, err, "error unmarshalling response body")
+	}
+
+	return responseData.ApiKey, nil
+}

api/server/handlers/project_oauth/upstash.go

@@ -0,0 +1,51 @@
+package project_oauth
+
+import (
+	"net/http"
+
+	"github.com/porter-dev/porter/internal/telemetry"
+
+	"golang.org/x/oauth2"
+
+	"github.com/porter-dev/porter/api/server/handlers"
+	"github.com/porter-dev/porter/api/server/shared"
+	"github.com/porter-dev/porter/api/server/shared/apierrors"
+	"github.com/porter-dev/porter/api/server/shared/config"
+	"github.com/porter-dev/porter/internal/oauth"
+)
+
+// ProjectOAuthUpstashHandler is the handler which redirects to the upstash oauth flow
+type ProjectOAuthUpstashHandler struct {
+	handlers.PorterHandlerReadWriter
+}
+
+// NewProjectOAuthUpstashHandler generates a new ProjectOAuthUpstashHandler
+func NewProjectOAuthUpstashHandler(
+	config *config.Config,
+	decoderValidator shared.RequestDecoderValidator,
+	writer shared.ResultWriter,
+) *ProjectOAuthUpstashHandler {
+	return &ProjectOAuthUpstashHandler{
+		PorterHandlerReadWriter: handlers.NewDefaultPorterHandler(config, decoderValidator, writer),
+	}
+}
+
+// ServeHTTP populates the oauth session with state and project id then redirects the user to the upstash oauth flow
+func (p *ProjectOAuthUpstashHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	ctx, span := telemetry.NewSpan(r.Context(), "serve-project-oauth-upstash")
+	defer span.End()
+
+	r = r.Clone(ctx)
+
+	state := oauth.CreateRandomState()
+
+	if err := p.PopulateOAuthSession(ctx, w, r, state, true, false, "", 0); err != nil {
+		err = telemetry.Error(ctx, span, err, "population oauth session failed")
+		p.HandleAPIError(w, r, apierrors.NewErrPassThroughToClient(err, http.StatusInternalServerError))
+		return
+	}
+
+	url := p.Config().UpstashConf.AuthCodeURL(state, oauth2.AccessTypeOffline, oauth2.SetAuthURLParam("audience", "upstash-api"))
+
+	http.Redirect(w, r, url, http.StatusFound)
+}

api/server/router/oauth_callback.go

@@ -51,6 +51,30 @@ func GetOAuthCallbackRoutes(
 		Router:   r,
 	})
 
+	// GET /api/oauth/upstash/callback -> oauth_callback.NewOAuthCallbackUpstashHandler
+	upstashEndpoint := factory.NewAPIEndpoint(
+		&types.APIRequestMetadata{
+			Verb:   types.APIVerbGet,
+			Method: types.HTTPVerbGet,
+			Path: &types.Path{
+				Parent:       basePath,
+				RelativePath: relPath + "/upstash/callback",
+			},
+		},
+	)
+
+	upstashHandler := oauth_callback.NewOAuthCallbackUpstashHandler(
+		config,
+		factory.GetDecoderValidator(),
+		factory.GetResultWriter(),
+	)
+
+	routes = append(routes, &router.Route{
+		Endpoint: upstashEndpoint,
+		Handler:  upstashHandler,
+		Router:   r,
+	})
+
 	// GET /api/oauth/digitalocean/callback -> oauth_callback.NewOAuthCallbackDOHandler
 	doEndpoint := factory.NewAPIEndpoint(
 		&types.APIRequestMetadata{

api/server/router/project_oauth.go

@@ -82,6 +82,34 @@ func getProjectOAuthRoutes(
 		Router:   r,
 	})
 
+	// GET /api/projects/{project_id}/oauth/upstash -> project_integration.NewProjectOAuthUpstashHandler
+	upstashEndpoint := factory.NewAPIEndpoint(
+		&types.APIRequestMetadata{
+			Verb:   types.APIVerbGet,
+			Method: types.HTTPVerbGet,
+			Path: &types.Path{
+				Parent:       basePath,
+				RelativePath: relPath + "/upstash",
+			},
+			Scopes: []types.PermissionScope{
+				types.UserScope,
+				types.ProjectScope,
+			},
+		},
+	)
+
+	upstashHandler := project_oauth.NewProjectOAuthUpstashHandler(
+		config,
+		factory.GetDecoderValidator(),
+		factory.GetResultWriter(),
+	)
+
+	routes = append(routes, &router.Route{
+		Endpoint: upstashEndpoint,
+		Handler:  upstashHandler,
+		Router:   r,
+	})
+
 	// GET /api/projects/{project_id}/oauth/digitalocean -> project_integration.NewProjectOAuthDOHandler
 	doEndpoint := factory.NewAPIEndpoint(
 		&types.APIRequestMetadata{

api/server/shared/config/config.go

@@ -76,6 +76,9 @@ type Config struct {
 	// SlackConf is the configuration for a Slack OAuth client
 	SlackConf *oauth2.Config
 
+	// UpstashConf is the configuration for an Upstash OAuth client
+	UpstashConf oauth2.Config
+
 	// WSUpgrader upgrades HTTP connections to websocket connections
 	WSUpgrader *websocket.Upgrader
 

api/server/shared/config/env/envconfs.go

@@ -82,6 +82,9 @@ type ServerConf struct {
 	SlackClientID     string `env:"SLACK_CLIENT_ID"`
 	SlackClientSecret string `env:"SLACK_CLIENT_SECRET"`
 
+	UpstashEnabled  bool   `env:"UPSTASH_ENABLED,default=false"`
+	UpstashClientID string `env:"UPSTASH_CLIENT_ID"`
+
 	BillingPrivateKey       string `env:"BILLING_PRIVATE_KEY"`
 	BillingPrivateServerURL string `env:"BILLING_PRIVATE_URL"`
 	BillingPublicServerURL  string `env:"BILLING_PUBLIC_URL"`

api/server/shared/config/loader/loader.go

@@ -261,6 +261,17 @@ func (e *EnvConfigLoader) LoadConfig() (res *config.Config, err error) {
 		res.Logger.Info().Msg("Created Slack client")
 	}
 
+	if sc.UpstashEnabled && sc.UpstashClientID != "" {
+		res.Logger.Info().Msg("Creating Upstash client")
+		res.UpstashConf = oauth.NewUpstashClient(&oauth.Config{
+			ClientID:     sc.UpstashClientID,
+			ClientSecret: "",
+			Scopes:       []string{"offline_access"},
+			BaseURL:      sc.ServerURL,
+		})
+		res.Logger.Info().Msg("Created Upstash client")
+	}
+
 	res.WSUpgrader = &websocket.Upgrader{
 		WSUpgrader: &gorillaws.Upgrader{
 			ReadBufferSize:  1024,

api/types/project_integration.go

@@ -8,6 +8,7 @@ const (
 	OAuthDigitalOcean OAuthIntegrationClient = "do"
 	OAuthGoogle       OAuthIntegrationClient = "google"
 	OAuthGitlab       OAuthIntegrationClient = "gitlab"
+	OAuthUpstash      OAuthIntegrationClient = "upstash"
 )
 
 // OAuthIntegrationClient is the name of an OAuth mechanism client

internal/models/integrations/upstash.go

@@ -0,0 +1,14 @@
+package integrations
+
+import "gorm.io/gorm"
+
+// UpstashIntegration is an integration for the Upstash service
+type UpstashIntegration struct {
+	gorm.Model
+
+	ProjectID uint `json:"project_id"`
+
+	SharedOAuthModel
+
+	DeveloperApiKey []byte `json:"developer_api_key"`
+}

internal/oauth/config.go

@@ -111,6 +111,20 @@ func NewSlackClient(cfg *Config) *oauth2.Config {
 	}
 }
 
+// NewUpstashClient creates a new oauth2.Config for Upstash
+func NewUpstashClient(cfg *Config) oauth2.Config {
+	return oauth2.Config{
+		ClientID:     cfg.ClientID,
+		ClientSecret: cfg.ClientSecret,
+		Endpoint: oauth2.Endpoint{
+			AuthURL:  "https://auth.upstash.com/authorize",
+			TokenURL: "https://auth.upstash.com/oauth/token",
+		},
+		RedirectURL: cfg.BaseURL + "/api/oauth/upstash/callback",
+		Scopes:      cfg.Scopes,
+	}
+}
+
 func CreateRandomState() string {
 	b := make([]byte, 16)
 	rand.Read(b)