// Package cibridge lets a runner on its owner's machine reach the CI
// engine through the forge (f24d78d9).
//
// dagnats already has an HTTP worker protocol — connect, poll, resolve,
// logs — authenticated by worker tokens scoped to task-type prefixes and
// one worker group (dagnats#627, #695). It listens only on the forge's
// box. This package forwards exactly those four routes, under a
// repository's URL, so a runner needs nothing but the forge's address,
// the repository's name and a token minted in Settings → Runners:
//
// POST /r/{repo}/ci/bridge/v1/workers/connect
// POST /r/{repo}/ci/bridge/v1/tasks/poll
// POST /r/{repo}/ci/bridge/v1/tasks/{id}/resolve
// POST /r/{repo}/ci/bridge/v1/tasks/{id}/logs
//
// A fifth route is the forge's own, not the engine's (RFC 0019
// §secrets):
//
// POST /r/{repo}/ci/bridge/v1/tasks/{id}/secrets
//
// A runner asks for the secrets a job it is running names. It proves it
// holds the task by the engine accepting a heartbeat for it with the
// runner's own token: dagnats lets only the token that claimed a task
// act on it (#627). What may be delivered is then the forge's decision
// alone, made from its own records (Deliver).
//
// Two things are decided here rather than trusted to the caller. A poll
// is pinned to the repository's own worker group, so a runner cannot
// ask for another repository's work by naming its group (the token
// would refuse it too; this makes the URL mean what it says). And the
// engine's ADMIN bearer is refused outright: on the box it is a
// convenience, on a public path it would be a master key.
package cibridge
import (
"bytes"
"crypto/subtle"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httputil"
"net/url"
"strings"
"time"
)
// maxBody bounds what a runner may send in one request. A log batch is
// the largest legitimate body; a runner sends many small ones.
const maxBody = 4 << 20
// Deliver answers a runner's request for secrets once it has proved it
// holds the task: the values by name, or an error saying why not. It
// must decide from the forge's own records, never from the request.
type Deliver func(repo, taskID, environment string, names []string) (map[string]string, error)
// Handler forwards the worker routes to engine. groupFor names a
// repository's worker group; known says whether the forge serves the
// repository at all; admin is the engine's admin bearer, refused here.
// deliver answers the secrets route; nil leaves the route out.
func Handler(engine string, admin string, groupFor func(repo string) string, known func(repo string) bool, deliver Deliver) (http.Handler, error) {
base, err := url.Parse(engine)
if err != nil || base.Scheme == "" || base.Host == "" {
return nil, fmt.Errorf("cibridge: engine URL %q is not an absolute URL", engine)
}
proxy := &httputil.ReverseProxy{
Rewrite: func(pr *httputil.ProxyRequest) {
pr.Out.URL.Scheme, pr.Out.URL.Host = base.Scheme, base.Host
pr.Out.Host = base.Host
// The forwarding headers name the runner's address to the
// engine, which neither needs nor logs it; drop them.
pr.Out.Header.Del("X-Forwarded-For")
},
// connect is a Server-Sent Events stream: every heartbeat must
// reach the runner as it happens, or the runner sees silence
// and the engine sees a dead worker.
FlushInterval: -1,
}
mux := http.NewServeMux()
route := func(pattern, upstream string, pin bool) {
mux.HandleFunc(pattern, func(w http.ResponseWriter, req *http.Request) {
repo := req.PathValue("repo")
if known != nil && !known(repo) {
http.Error(w, "no such repository", http.StatusNotFound)
return
}
tok, ok := bearer(req)
if !ok {
w.Header().Set("WWW-Authenticate", `Bearer realm="quarry runners"`)
http.Error(w, "a runner authenticates with a worker token from Settings → Runners", http.StatusUnauthorized)
return
}
if admin != "" && subtle.ConstantTimeCompare([]byte(tok), []byte(admin)) == 1 {
http.Error(w, "the engine's admin token is not accepted here; mint a worker token in Settings → Runners", http.StatusForbidden)
return
}
req.Body = http.MaxBytesReader(w, req.Body, maxBody)
if pin {
body, err := io.ReadAll(req.Body)
if err != nil {
http.Error(w, "request too large", http.StatusRequestEntityTooLarge)
return
}
pinned, err := pinGroup(body, groupFor(repo))
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
req.Body = io.NopCloser(bytes.NewReader(pinned))
req.ContentLength = int64(len(pinned))
}
path := upstream
if id := req.PathValue("id"); id != "" {
path = strings.Replace(upstream, "{id}", url.PathEscape(id), 1)
}
req.URL.Path, req.URL.RawPath = path, ""
proxy.ServeHTTP(w, req)
})
}
route("POST /r/{repo}/ci/bridge/v1/workers/connect", "/v1/workers/connect", true)
route("POST /r/{repo}/ci/bridge/v1/tasks/poll", "/v1/tasks/poll", true)
route("POST /r/{repo}/ci/bridge/v1/tasks/{id}/resolve", "/v1/tasks/{id}/resolve", false)
route("POST /r/{repo}/ci/bridge/v1/tasks/{id}/logs", "/v1/tasks/{id}/logs", false)
if deliver != nil {
mux.HandleFunc("POST /r/{repo}/ci/bridge/v1/tasks/{id}/secrets", secretsHandler(base, admin, known, deliver))
}
return mux, nil
}
// secretsBody is a runner's request: the job's environment ("" for
// none) and the secret names its steps export.
type secretsBody struct {
Environment string `json:"environment"`
Names []string `json:"names"`
}
func secretsHandler(base *url.URL, admin string, known func(string) bool, deliver Deliver) http.HandlerFunc {
client := &http.Client{Timeout: 15 * time.Second}
return func(w http.ResponseWriter, req *http.Request) {
w.Header().Set("Cache-Control", "no-store")
repo, id := req.PathValue("repo"), req.PathValue("id")
if known != nil && !known(repo) {
http.Error(w, "no such repository", http.StatusNotFound)
return
}
tok, ok := bearer(req)
if !ok {
w.Header().Set("WWW-Authenticate", `Bearer realm="quarry runners"`)
http.Error(w, "a runner authenticates with a worker token from Settings → Runners", http.StatusUnauthorized)
return
}
if admin != "" && subtle.ConstantTimeCompare([]byte(tok), []byte(admin)) == 1 {
http.Error(w, "the engine's admin token is not accepted here", http.StatusForbidden)
return
}
var body secretsBody
if err := json.NewDecoder(http.MaxBytesReader(w, req.Body, 64<<10)).Decode(&body); err != nil || len(body.Names) == 0 || len(body.Names) > 100 {
http.Error(w, "the body is {\"environment\": \"...\", \"names\": [...]} with 1 to 100 names", http.StatusBadRequest)
return
}
// The proof of claim: only the token that claimed this task can
// heartbeat it, and a heartbeat changes nothing but its lease.
hb, _ := http.NewRequestWithContext(req.Context(), "POST",
base.Scheme+"://"+base.Host+"/v1/tasks/"+url.PathEscape(id)+"/resolve",
strings.NewReader(`{"action":"heartbeat"}`))
hb.Header.Set("Authorization", "Bearer "+tok)
hb.Header.Set("Content-Type", "application/json")
resp, err := client.Do(hb)
if err != nil {
http.Error(w, "the engine could not be asked whether this runner holds the task", http.StatusBadGateway)
return
}
io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10))
resp.Body.Close()
if resp.StatusCode >= 300 {
http.Error(w, "this runner does not hold that task", http.StatusForbidden)
return
}
values, err := deliver(repo, id, body.Environment, body.Names)
if err != nil {
http.Error(w, err.Error(), http.StatusForbidden)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]map[string]string{"secrets": values})
}
}
func bearer(req *http.Request) (string, bool) {
h := req.Header.Get("Authorization")
tok, ok := strings.CutPrefix(h, "Bearer ")
tok = strings.TrimSpace(tok)
return tok, ok && tok != ""
}
// pinGroup sets a poll's or a connect's worker_group to the repository's own, and
// refuses one that names another. Every other field passes through
// untouched.
func pinGroup(body []byte, own string) ([]byte, error) {
var poll map[string]json.RawMessage
if len(bytes.TrimSpace(body)) == 0 {
poll = map[string]json.RawMessage{}
} else if err := json.Unmarshal(body, &poll); err != nil {
return nil, fmt.Errorf("a poll is a JSON object: %v", err)
}
if raw, ok := poll["worker_group"]; ok {
var named string
if err := json.Unmarshal(raw, &named); err != nil {
return nil, fmt.Errorf("worker_group must be a string")
}
if named != "" && named != own {
return nil, fmt.Errorf("this repository's runners poll worker group %q, not %q", own, named)
}
}
g, _ := json.Marshal(own)
poll["worker_group"] = g
return json.Marshal(poll)
}