From 8bd6005b06d0b8cea57964120a3e6096ded69b82 Mon Sep 17 00:00:00 2001 From: Drew DeVault Date: Tue, 17 Aug 2021 14:05:55 +0200 Subject: [PATCH] webhooks: initial prototype for GQL-native webhooks --- auth/bearer.go | 35 +++++ auth/middleware.go | 58 +++++---- server/directives.go | 6 +- webhooks/config.go | 46 +++++++ webhooks/context.go | 39 ++++++ webhooks/queue.go | 297 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 453 insertions(+), 28 deletions(-) create mode 100644 webhooks/config.go create mode 100644 webhooks/context.go create mode 100644 webhooks/queue.go diff --git a/auth/bearer.go b/auth/bearer.go index 8ea49df04b0a35f937ed0dd3d8c22e21eb27cb43..940f9d5276261f26ea6027fd990b61df6fdf4590 100644 --- a/auth/bearer.go +++ b/auth/bearer.go @@ -1,13 +1,17 @@ package auth import ( + "context" "encoding/base64" "encoding/hex" + "fmt" "log" + "strings" "time" "git.sr.ht/~sircmpwn/go-bare" + "git.sr.ht/~sircmpwn/core-go/config" "git.sr.ht/~sircmpwn/core-go/crypto" ) @@ -75,3 +79,34 @@ func DecodeBearerToken(token string) *BearerToken { } return &bt } + +func DecodeGrants(ctx context.Context, grants string) map[string]string { + if grants == "" { + // All permissions + return nil + } + accessMap := make(map[string]string) + for _, grant := range strings.Split(grants, " ") { + var ( + service string + scope string + access string + ) + parts := strings.Split(grant, "/") + if len(parts) != 2 { + panic(fmt.Errorf("OAuth grant '%s' without service/scope format", grant)) + } + service = parts[0] + parts = strings.Split(parts[1], ":") + scope = parts[0] + if len(parts) == 1 { + access = "RO" + } else { + access = parts[1] + } + if service == config.ServiceName(ctx) { + accessMap[scope] = access + } + } + return accessMap +} diff --git a/auth/middleware.go b/auth/middleware.go index 038a00ea6e11decbdb36d2685d6bdd715bd2943c..a7bc2e1df3b03f914ed04f2da88942031cc02ed4 100644 --- a/auth/middleware.go +++ b/auth/middleware.go @@ -53,6 +53,7 @@ const ( AUTH_OAUTH2 = iota AUTH_COOKIE = iota AUTH_INTERNAL = iota + AUTH_WEBHOOK = iota ) type AuthContext struct { @@ -74,9 +75,10 @@ type AuthContext struct { // Only filled out if AuthMethod == AUTH_INTERNAL InternalAuth InternalAuth - // Only filled out if AuthMethod == AUTH_OAUTH2 + // Only filled out if AuthMethod == AUTH_OAUTH2 or AUTH_WEBHOOK BearerToken *BearerToken Access map[string]string + TokenHash [64]byte } func authError(w http.ResponseWriter, reason string, code int) { @@ -551,32 +553,8 @@ func OAuth2(token string, hash [64]byte, w http.ResponseWriter, auth.AuthMethod = AUTH_OAUTH2 auth.BearerToken = bt - - if bt.Grants != "" { - auth.Access = make(map[string]string) - for _, grant := range strings.Split(bt.Grants, " ") { - var ( - service string - scope string - access string - ) - parts := strings.Split(grant, "/") - if len(parts) != 2 { - panic(fmt.Errorf("OAuth grant '%s' without service/scope format", grant)) - } - service = parts[0] - parts = strings.Split(parts[1], ":") - scope = parts[0] - if len(parts) == 1 { - access = "RO" - } else { - access = parts[1] - } - if service == config.ServiceName(r.Context()) { - auth.Access[scope] = access - } - } - } + auth.TokenHash = hash + auth.Access = DecodeGrants(r.Context(), bt.Grants) ctx := context.WithValue(r.Context(), userCtxKey, &auth) r = r.WithContext(ctx) @@ -671,6 +649,32 @@ func LegacyOAuth(bearer string, hash [64]byte, w http.ResponseWriter, next.ServeHTTP(w, r) } +// Returns an auth context configured for webhook delivery. This auth +// configuration is not possible during a normal GraphQL query, and is only +// used during webhook execution. +// +// The "ctx" parameter should be a webhook context, and the "auth" parameter +// should be the authentication context from the request which caused the +// webhook to be fired. +func WebhookAuth(ctx context.Context, auth *AuthContext, + tokenHash [64]byte, grants string, clientID *string, + expires time.Time) (context.Context, error) { + if time.Now().UTC().After(expires) { + return nil, fmt.Errorf("The authentication token used to create this webhook has expired") + } + + whAuth := *auth + whAuth.AuthMethod = AUTH_WEBHOOK + whAuth.TokenHash = tokenHash + whAuth.Access = DecodeGrants(ctx, grants) + whAuth.BearerToken = &BearerToken{} + if clientID != nil { + whAuth.BearerToken.ClientID = *clientID + } + + return context.WithValue(ctx, userCtxKey, &whAuth), nil +} + func Middleware(conf ini.File, apiconf string) func(http.Handler) http.Handler { var internalNet []*net.IPNet src, ok := conf.Get(apiconf, "internal-ipnet") diff --git a/server/directives.go b/server/directives.go index 4a95751e17c7869ba8f55776d4e1f01dff08a5b3..8aa2399f08810cb5e8ca127add0fe127adc4a4cc 100644 --- a/server/directives.go +++ b/server/directives.go @@ -21,7 +21,6 @@ func Internal(ctx context.Context, obj interface{}, func Access(ctx context.Context, obj interface{}, next graphql.Resolver, scope string, kind string) (interface{}, error) { - authctx := auth.ForContext(ctx) switch authctx.AuthMethod { @@ -32,6 +31,11 @@ func Access(ctx context.Context, obj interface{}, next graphql.Resolver, // Only legacy tokens with "*" scopes ever get this far return next(ctx) } + case auth.AUTH_WEBHOOK: + if kind != "RO" { + return nil, fmt.Errorf("Access to read/write resolver denied for webhook") + } + fallthrough case auth.AUTH_OAUTH2: if authctx.Access == nil { return next(ctx) diff --git a/webhooks/config.go b/webhooks/config.go new file mode 100644 index 0000000000000000000000000000000000000000..d35f07e09904414d354bfcd1f4cf2ac576ba261a --- /dev/null +++ b/webhooks/config.go @@ -0,0 +1,46 @@ +package webhooks + +import ( + "context" + "encoding/hex" + "fmt" + "time" + + "git.sr.ht/~sircmpwn/core-go/auth" +) + +type AuthConfig struct { + TokenHash string + Grants string + ClientID *string + Expires time.Time +} + +// Pulls auth details out of the config context and returns a structure of all +// of the information necessary to build a webhook context with the same +// authentication parameters. +func NewAuthConfig(ctx context.Context) (AuthConfig, error) { + user := auth.ForContext(ctx) + switch user.AuthMethod { + case auth.AUTH_OAUTH_LEGACY: + return AuthConfig{}, fmt.Errorf("Native webhooks are not supported with legacy OAuth") + case auth.AUTH_OAUTH2: + ac := AuthConfig{ + TokenHash: hex.EncodeToString(user.TokenHash[:]), + Grants: user.BearerToken.Grants, + Expires: user.BearerToken.Expires.Time(), + } + if user.BearerToken.ClientID != "" { + clientID := user.BearerToken.ClientID + ac.ClientID = &clientID + } + return ac, nil + case auth.AUTH_COOKIE: + // TODO: Should this work? + return AuthConfig{}, fmt.Errorf("Native webhooks are not supported with web authentication") + case auth.AUTH_INTERNAL: + // TODO: Should this work? + panic("Internal webtoken auth is not supported") + } + panic("Unreachable") +} diff --git a/webhooks/context.go b/webhooks/context.go new file mode 100644 index 0000000000000000000000000000000000000000..e4ea66ce7f5aa17f37fa3d2478f6245dc3686bc0 --- /dev/null +++ b/webhooks/context.go @@ -0,0 +1,39 @@ +package webhooks + +import ( + "context" + "errors" + + "github.com/google/uuid" + + "git.sr.ht/~sircmpwn/core-go/auth" +) + +type contextKey struct { + name string +} + +var payloadContextKey = &contextKey{"webhookPayloadContext"} + +type WebhookContext struct { + Name string + Event string + User *auth.AuthContext + Payload interface{} + PayloadUUID uuid.UUID + Subscription *WebhookSubscription +} + +// Prepares an context for a specific webhook delivery. +func Context(ctx context.Context, payload interface{}) context.Context { + return context.WithValue(ctx, payloadContextKey, payload) +} + +// Returns the active payload for a webhook context. +func Payload(ctx context.Context) (interface{}, error) { + payload := ctx.Value(payloadContextKey) + if payload == nil { + return nil, errors.New("Cannot use this resolver without an active webhook context") + } + return payload, nil +} diff --git a/webhooks/queue.go b/webhooks/queue.go new file mode 100644 index 0000000000000000000000000000000000000000..b3652b8ceded141945fd585d14e498573da01923 --- /dev/null +++ b/webhooks/queue.go @@ -0,0 +1,297 @@ +package webhooks + +import ( + "bytes" + "context" + "database/sql" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "io/ioutil" + "log" + "net/http" + "strings" + "time" + + "git.sr.ht/~sircmpwn/dowork" + "github.com/99designs/gqlgen/graphql" + "github.com/99designs/gqlgen/graphql/executor" + "github.com/google/uuid" + sq "github.com/Masterminds/squirrel" + + "git.sr.ht/~sircmpwn/core-go/auth" + "git.sr.ht/~sircmpwn/core-go/crypto" + "git.sr.ht/~sircmpwn/core-go/database" +) + +type WebhookQueue struct { + Queue *work.Queue + Schema graphql.ExecutableSchema +} + +type WebhookSubscription struct { + ID int + URL string + Query string + TokenHash string + Grants string + ClientID *string + Expires time.Time +} + +// Creates a new worker for delivering webhooks. The caller must start the +// worker themselves. +func NewQueue(schema graphql.ExecutableSchema) *WebhookQueue { + return &WebhookQueue{work.NewQueue("webhooks"), schema} +} + +// Schedules delivery of a webhook to a set of subscribers. +// +// The select builder should not return any columns, i.e. the caller should use +// squirrel.Select() with no parameters. The caller should prepare FROM and any +// WHERE clauses which are necessary to refine the subscriber list (e.g. by +// affected resource ID). The caller must alias the webhook table to "sub", e.g. +// sq.Select().From("my_webhook_subscription sub"). +// +// Name shall be the prefix of the webhook tables, e.g. "profile" for +// "gql_profile_wh_{delivery,sub}". +// +// The context should NOT be the context used to service the HTTP request which +// initiated the webhook delivery. It should instead be a fresh background +// context which contains the necessary state for your application to process +// the webhook resolvers. +func (queue *WebhookQueue) Schedule(ctx context.Context, q sq.SelectBuilder, + name, event string, payloadUUID uuid.UUID, payload interface{}) { + user := auth.ForContext(ctx) + // The following tasks are done during this process: + // + // 1. Fetch subscription details from the database + // 2. Prepare deliveries and create delivery records + // 3. Deliver the webhooks + // + // The first two steps are done in this task, then N tasks are created for + // step 3 where N = number of subscriptions. + task := work.NewTask(func(ctx context.Context) error { + ctx = Context(ctx, payload) + subs, err := queue.fetchSubscriptions(ctx, q, event) + if err != nil { + return err + } + if len(subs) == 0 { + return nil + } + + tasks := make([]*work.Task, len(subs)) + if err := database.WithTx(ctx, nil, func(tx *sql.Tx) error { + var err error + for i, sub := range subs { + webhook := WebhookContext{ + Name: name, + Event: event, + User: user, + Payload: payload, + PayloadUUID: payloadUUID, + Subscription: sub, + } + tasks[i], err = queue.queueStage2(ctx, tx, &webhook) + if err != nil { + return err + } + } + return nil + }); err != nil { + log.Printf("Failed to enqueue %s/%s webhooks: %v", event, err) + return err + } + + for _, task := range tasks { + queue.Queue.Enqueue(task) + } + log.Printf("Enqueued %s/%s webhook delivery for %d subscriptions", + name, event, len(subs)) + return nil + }) + queue.Queue.Enqueue(task) +} + +func (queue *WebhookQueue) fetchSubscriptions(ctx context.Context, + q sq.SelectBuilder, event string) ([]*WebhookSubscription, error) { + var subs []*WebhookSubscription + if err := database.WithTx(ctx, &sql.TxOptions{ + Isolation: 0, + ReadOnly: true, + }, func(tx *sql.Tx) error { + var ( + err error + rows *sql.Rows + ) + if rows, err = q. + Columns("sub.id", "sub.url", "sub.query", + "sub.token_hash", "sub.grants", "sub.client_id", + "sub.expires"). + Where("? = ANY(sub.events)", event). + PlaceholderFormat(sq.Dollar). + RunWith(tx). + QueryContext(ctx); err != nil { + return err + } + defer rows.Close() + + for rows.Next() { + var sub WebhookSubscription + if err := rows.Scan(&sub.ID, &sub.URL, &sub.Query, + &sub.TokenHash, &sub.Grants, &sub.ClientID, + &sub.Expires); err != nil { + panic(err) + } + subs = append(subs, &sub) + } + + return nil + }); err != nil { + return nil, err + } + return subs, nil +} + +func (queue *WebhookQueue) queueStage2(ctx context.Context, + tx *sql.Tx, webhook *WebhookContext) (*work.Task, error) { + headers := make(http.Header) + headers.Set("Content-Type", "application/json") + headers.Set("X-Webhook-Event", webhook.Event) + headers.Set("X-Webhook-Delivery", webhook.PayloadUUID.String()) + + sub := webhook.Subscription + tslice, err := hex.DecodeString(sub.TokenHash) + if err != nil { + panic(err) + } + + var tokenHash [64]byte + copy(tokenHash[:], tslice) + ctx, err = auth.WebhookAuth(ctx, webhook.User, + tokenHash, sub.Grants, sub.ClientID, sub.Expires) + if err != nil { + // TODO: This codepath can occur when the token has expired, and we may + // want to communicate this to the user. + return nil, err + } + + exec := executor.New(queue.Schema) + params := graphql.RawParams{ + Query: sub.Query, + ReadTime: graphql.TraceTiming{ + Start: graphql.Now(), + End: graphql.Now(), + }, + } + ctx = graphql.StartOperationTrace(ctx) + rc, errors := exec.CreateOperationContext(ctx, ¶ms) + if errors != nil { + panic(errors) + } + ctx = graphql.WithOperationContext(ctx, rc) + var resp graphql.ResponseHandler + resp, ctx = exec.DispatchOperation(ctx, rc) + payload, err := json.Marshal(resp(ctx)) + if err != nil { + panic(err) + } + + var deliveryID int + err = sq. + Insert("gql_"+webhook.Name+"_wh_delivery"). + Columns("uuid", "date", "event", "subscription_id", "request_body"). + Values(webhook.PayloadUUID, sq.Expr("NOW() at time zone 'utc'"), + webhook.Event, sub.ID, string(payload)). + Suffix(`RETURNING (id)`). + PlaceholderFormat(sq.Dollar). + RunWith(tx). + ScanContext(ctx, &deliveryID) + if err != nil { + return nil, err + } + + return work.NewTask(func(ctx context.Context) error { + queue.deliverPayload(ctx, webhook, headers, payload, deliveryID) + return nil + }).Retries(5).After(func(ctx context.Context, task *work.Task) { + if task.Result() == nil { + log.Printf("%s: webhook delivery complete after %d attempts", + webhook.PayloadUUID, task.Attempts()) + } else { + log.Printf("%s: webhook delivery failed after %d attempts: %v", + webhook.PayloadUUID, task.Attempts(), task.Result()) + } + }), nil +} + +// Performs a webhook delivery and updates the delivery record in the database +func (queue *WebhookQueue) deliverPayload(ctx context.Context, + webhook *WebhookContext, headers http.Header, payload []byte, + deliveryID int) error { + + client := &http.Client{ + Timeout: 30 * time.Second, + } + rctx, cancel := context.WithDeadline(ctx, time.Now().Add(30*time.Second)) + req, err := http.NewRequestWithContext(rctx, + http.MethodPost, webhook.Subscription.URL, bytes.NewReader(payload)) + defer cancel() + if err != nil { + return fmt.Errorf("http.NewRequestWithContext: %v: %e", + err, work.ErrDoNotReattempt) + } + + req.Header = make(http.Header) + for key, values := range headers { + for _, value := range values { + req.Header.Add(key, value) + } + } + nonce, sig := crypto.SignWebhook(payload) + req.Header.Add("X-Payload-Nonce", nonce) + req.Header.Add("X-Payload-Signature", sig) + + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + reader := io.LimitReader(resp.Body, 262144) // No more than 256 KiB + body, err := ioutil.ReadAll(reader) + if err != nil { + return fmt.Errorf("Error reading response body: %v: %e", + err, work.ErrDoNotReattempt) + } + + if err = database.WithTx(ctx, nil, func(tx *sql.Tx) error { + var theirs strings.Builder + resp.Header.Write(&theirs) + _, err := sq. + Update("gql_"+webhook.Name+"_wh_delivery"). + Set("response_body", string(body)). + Set("response_status", resp.StatusCode). + Set("response_headers", theirs.String()). + Where("id = ?", deliveryID). + PlaceholderFormat(sq.Dollar). + RunWith(tx). + ExecContext(ctx) + return err + }); err != nil { + log.Printf("Warning: webhook delivered, but updating delivery record failed: %v", err) + return nil + } + + if resp.StatusCode == http.StatusBadGateway || + resp.StatusCode == http.StatusServiceUnavailable || + resp.StatusCode == http.StatusGatewayTimeout { + // Retry + return fmt.Errorf("Server returned status %d: %s", + resp.StatusCode, resp.Status) + } + + return nil +}