// 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
//
// 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"
)
// 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
// 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.
func Handler(engine string, admin string, groupFor func(repo string) string, known func(repo string) bool) (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", false)
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)
return mux, nil
}
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 worker_group to the repository's own, and
// refuses a poll 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)
}