diff --git a/README.md b/README.md index 847b079..b50c3ba 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,8 @@ where every job runs in a fresh virtual machine. > **Status:** early development. A pipeline runs end to end: the controller schedules its jobs, a worker checks > out the exact commit and runs each job's commands through the agent with the secrets the admin allows, and -> the output is stored. Real VMs and Git providers come next; see [docs/ROADMAP.md](docs/ROADMAP.md). +> the output is stored. Runs start from a `git push`, to your own git server or to GitHub. Real VMs and +> Opifica-hosted repositories come next; see [docs/ROADMAP.md](docs/ROADMAP.md). > See the [organisation page](https://github.com/opifica) for the architecture. ## What lives here @@ -14,7 +15,7 @@ where every job runs in a fresh virtual machine. |---|---|---| | `internal/pipeline` | The pipeline language: YAML parser, validation, CEL conditions, matrix expansion, planning | Working | | `schema/` | JSON Schema for pipeline files, for editor autocompletion | Working | -| `cmd/opifica` | CLI: `validate`, `plan`, `schema` | Working | +| `cmd/opifica` | CLI: `validate`, `plan`, `schema`, `trigger`, `hook` | Working | | `proto/` | API definitions: worker ↔ controller, worker ↔ guest agent (Protobuf, ConnectRPC) | Draft | | `pkg/worker` | Shared worker kit: `Hypervisor` interface, warm VM pool, controller client, Git mirrors, job runner, `worker.Run` | Working | | `pkg/worker/workertest` | Fake hypervisor for testing worker code without VMs | Working | @@ -22,7 +23,9 @@ where every job runs in a fresh virtual machine. | `internal/controller/scheduler` | Runs and jobs, matching jobs to workers, leases, retries, cancellation | Working | | `internal/controller/logs` | Job output on the controller's disk | Working | | `internal/controller/secrets` | Encrypted secrets, handed only to jobs the admin's policy allows; every request audited | Working | -| | Git providers, admin API | Not started | +| `internal/controller/repos` | Repositories, routing rules (event + ref → pool), trigger tokens; the trigger API | Working | +| `cmd/opifica` `trigger`, `hook` | Start runs from GitHub Actions, or from the git hook of a bare repository | Working | +| | Hosted repositories (opifica-git), admin API | Not started | | `cmd/agent`, `internal/agent` | Runs inside each VM: checkout, running steps, masking secrets in output, killing process trees on cancel and timeout | Working | | `cmd/devworker` | A worker without VMs for trying Opifica on one machine (no isolation) | Working | | `cmd/signer` | Signing service for release artifacts | Not started | @@ -140,13 +143,53 @@ bin/controller serve -addr 127.0.0.1:8443 & # TLS; keep it running bin/devworker join -controller https://127.0.0.1:8443 -token opj_... -ca-fingerprint sha256:... bin/devworker run -controller https://127.0.0.1:8443 & -# Until Git providers exist, start runs by hand: -bin/controller run create -repo acme/app -clone-url "$PWD" -pool trusted \ - -sha "$(git rev-parse HEAD)" -file .opifica/pipeline.yaml -bin/controller run show +# Register a repository; prints its trigger token +bin/controller repo create acme/app -clone-url "$PWD" -route push:main=trusted -route '*=untrusted' + +# Start a run for the current branch and commit, and follow it +export OPIFICA_CONTROLLER=https://127.0.0.1:8443 OPIFICA_CA_FINGERPRINT=sha256:... \ + OPIFICA_REPO=acme/app OPIFICA_TRIGGER_TOKEN=opr_... +bin/opifica trigger -wait + +bin/controller run list bin/controller run logs ``` +The controller needs `-allow-local-repos` to build from a local path like `$PWD`. + +### Triggering runs + +A run starts when a ref moves: a push to a branch, a tag, or a pull request. The controller reads +`.opifica/pipeline.yaml` at that commit itself, and the repository's **routes** decide the pool. The first matching +route wins, and an event no route matches starts nothing: + +```sh +bin/controller repo create acme/app -clone-url ssh://git@git.example/acme/app.git \ + -route push:main=trusted -route tag:v*=release -route pull_request=untrusted +``` + +**Your own git server.** Install the hook in the bare repository as `hooks/post-receive`: + +```sh +#!/bin/sh +export OPIFICA_CONTROLLER=https://ci.example:8443 OPIFICA_CA_FINGERPRINT=sha256:... OPIFICA_REPO=acme/app +export OPIFICA_TRIGGER_TOKEN="$(cat /etc/opifica/acme-app.token)" +exec opifica hook post-receive +``` + +Each push then prints the runs it started: + +```console +$ git push origin main +remote: opifica: main: run 0b6f… started in pool trusted (3 jobs) +``` + +Opifica-hosted repositories, where creating a repository sets this up for you, are on the roadmap. + +**GitHub.** Add [`examples/github-actions.yml`](examples/github-actions.yml) as a workflow. Its step runs +`opifica trigger -wait`, which passes or fails with the Opifica run and prints its output. For a private +repository, register it with `-credential-stdin` and a read-only token. + ### Secrets The admin decides who gets a secret; a pipeline can only ask for it. The value is read from standard input: diff --git a/cmd/controller/main.go b/cmd/controller/main.go index 5536454..14ffed7 100644 --- a/cmd/controller/main.go +++ b/cmd/controller/main.go @@ -1,9 +1,10 @@ -// Command controller runs the Opifica controller: the API, Git provider -// integration, pipeline engine, scheduler and secrets broker. +// Command controller runs the Opifica controller: the API, the repository +// registry and trigger API, pipeline engine, scheduler and secrets broker. // // Implemented so far: the database, the built-in certificate authority, join // tokens, Join, the Connect stream, the scheduler (runs, jobs, leases, -// retries, cancellation), job logs and secrets. Git providers come next. +// retries, cancellation), job logs, secrets, and repositories with routing +// and triggers. package main import ( @@ -23,13 +24,16 @@ import ( "github.com/google/uuid" + opificav1 "github.com/opifica/opifica/gen/opifica/v1" "github.com/opifica/opifica/internal/controller/api" "github.com/opifica/opifica/internal/controller/ca" "github.com/opifica/opifica/internal/controller/logs" + "github.com/opifica/opifica/internal/controller/repos" "github.com/opifica/opifica/internal/controller/scheduler" "github.com/opifica/opifica/internal/controller/secrets" "github.com/opifica/opifica/internal/controller/store" "github.com/opifica/opifica/internal/controller/workers" + "github.com/opifica/opifica/internal/gitmirror" "github.com/opifica/opifica/internal/server" "github.com/opifica/opifica/internal/version" ) @@ -41,7 +45,11 @@ Commands: token create Create a one-time join token for a new build machine worker list List workers worker revoke Revoke a worker; it is disconnected and can't reconnect - run create Start a run from a pipeline file (until Git providers exist) + repo create Register a repository with its routing rules; prints a trigger token + repo list List repositories + repo delete Delete a repository + repo token create Create another trigger token for a repository + run create Start a run from a local pipeline file (for testing) run list List recent runs run show Show a run and its jobs run cancel Cancel a run @@ -83,6 +91,8 @@ func runWithInput(args []string, stdin io.Reader, stdout, stderr io.Writer) int return runCommand(args[1], args[2:], stdout, stderr) case args[0] == "secret" && len(args) > 1: return secretCommand(args[1], args[2:], stdin, stdout, stderr) + case args[0] == "repo" && len(args) > 1: + return repoCommand(args[1], args[2:], stdin, stdout, stderr) case args[0] == "version": fmt.Fprintln(stdout, version.Version) return 0 @@ -130,6 +140,7 @@ func serve(args []string, stderr io.Writer) int { var c common c.register(fs) addr := fs.String("addr", ":8443", "address to listen on") + allowLocal := fs.Bool("allow-local-repos", false, "allow repositories whose clone URL is a local path (hosted on this machine, or development)") hosts := fs.String("hosts", defaultHosts(), "comma-separated names and IPs workers use to reach the controller") if err := fs.Parse(args); err != nil { return 2 @@ -158,13 +169,23 @@ func serve(args []string, stderr io.Writer) int { log.Error("loading secrets key", "err", err) return 1 } + logStore := &logs.Store{Dir: filepath.Join(c.dataDir, "logs")} + var reposSvc *repos.Service sched := scheduler.New(svc.Store, svc.Sessions(), log.With("component", "scheduler"), scheduler.Config{ - Logs: &logs.Store{Dir: filepath.Join(c.dataDir, "logs")}, + Logs: logStore, Secrets: &secrets.Manager{Store: svc.Store, Key: key}, + Credentials: func(ctx context.Context, run uuid.UUID) (*opificav1.Credential, error) { + return reposSvc.RunCredential(ctx, run) + }, }) + reposSvc = &repos.Service{ + Store: svc.Store, Key: key, Scheduler: sched, + Mirrors: &gitmirror.Mirrors{Dir: filepath.Join(c.dataDir, "mirrors"), AllowLocal: *allowLocal}, + } svc.Messages = sched go sched.Run(ctx) - if err := server.RunTLS(ctx, *addr, api.NewHandler(svc), tlsCfg, log); err != nil { + handler := api.NewHandler(api.Deps{Workers: svc, Repos: reposSvc, Logs: logStore}) + if err := server.RunTLS(ctx, *addr, handler, tlsCfg, log); err != nil { log.Error("server stopped", "err", err) return 1 } diff --git a/cmd/controller/main_test.go b/cmd/controller/main_test.go index b90ce61..2b6249b 100644 --- a/cmd/controller/main_test.go +++ b/cmd/controller/main_test.go @@ -66,7 +66,7 @@ func TestTokenCreateAndJoinOverTLS(t *testing.T) { if err != nil { t.Fatal(err) } - go server.ServeTLS(ctx, ln, api.NewHandler(svc), tlsCfg, svc.Log) + go server.ServeTLS(ctx, ln, api.NewHandler(api.Deps{Workers: svc}), tlsCfg, svc.Log) // The worker has no CA yet: it accepts the controller only if the // certificate chains to a CA with the pinned fingerprint. diff --git a/cmd/controller/repos.go b/cmd/controller/repos.go new file mode 100644 index 0000000..2b56be4 --- /dev/null +++ b/cmd/controller/repos.go @@ -0,0 +1,204 @@ +package main + +import ( + "bytes" + "context" + "flag" + "fmt" + "io" + "log/slog" + "path/filepath" + "strings" + "text/tabwriter" + "time" + + "github.com/opifica/opifica/internal/controller/repos" + "github.com/opifica/opifica/internal/controller/scheduler" + "github.com/opifica/opifica/internal/controller/secrets" + "github.com/opifica/opifica/internal/gitmirror" +) + +func repoCommand(sub string, args []string, stdin io.Reader, stdout, stderr io.Writer) int { + switch sub { + case "create": + return repoCreate(args, stdin, stdout, stderr) + case "list": + return repoList(args, stdout, stderr) + case "delete": + return repoDelete(args, stdout, stderr) + case "token": + if len(args) > 0 && args[0] == "create" { + return repoTokenCreate(args[1:], stdout, stderr) + } + } + fmt.Fprintf(stderr, "controller: unknown command \"repo %s\"\n\n%s", sub, usage) + return 2 +} + +func withRepos(c common, stderr io.Writer, fn func(ctx context.Context, r *repos.Service) error) int { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + svc, closeFn, err := c.open(ctx) + if err != nil { + fmt.Fprintf(stderr, "controller: %v\n", err) + return 1 + } + defer closeFn() + key, err := secrets.LoadOrCreateKey(c.dataDir) + if err != nil { + fmt.Fprintf(stderr, "controller: loading secrets key: %v\n", err) + return 1 + } + quiet := slog.New(slog.NewTextHandler(io.Discard, nil)) + r := &repos.Service{ + Store: svc.Store, Key: key, + Scheduler: scheduler.New(svc.Store, svc.Sessions(), quiet, scheduler.Config{}), + Mirrors: &gitmirror.Mirrors{Dir: filepath.Join(c.dataDir, "mirrors")}, + } + if err := fn(ctx, r); err != nil { + fmt.Fprintf(stderr, "controller: %v\n", err) + return 1 + } + return 0 +} + +func repoCreate(args []string, stdin io.Reader, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("repo create", flag.ContinueOnError) + fs.SetOutput(stderr) + var c common + c.register(fs) + cloneURL := fs.String("clone-url", "", "where the controller and workers fetch the code (required)") + var routes listFlag + fs.Var(&routes, "route", "event[:ref]=pool, first match wins (repeatable; default *=untrusted), e.g. push:main=trusted") + credStdin := fs.Bool("credential-stdin", false, "read a read-only clone token from standard input") + fs.Usage = func() { + fmt.Fprint(stderr, `Usage: controller repo create [options] OWNER/NAME + +Registers a repository and prints its first trigger token. + +Routes decide which pool runs each event; the first matching route wins and +an event no route matches starts no run. For example: + + -route push:main=trusted -route tag:v*=release -route '*=untrusted' + +Options: +`) + fs.PrintDefaults() + } + pos, err := parseWithArgs(fs, args) + if err != nil { + return 2 + } + if len(pos) != 1 || *cloneURL == "" { + fs.Usage() + return 2 + } + var parsed []repos.Route + for _, r := range routes { + route, err := repos.ParseRoute(r) + if err != nil { + fmt.Fprintf(stderr, "controller: %v\n", err) + return 2 + } + parsed = append(parsed, route) + } + var cred []byte + if *credStdin { + b, err := io.ReadAll(io.LimitReader(stdin, 16<<10)) + if err != nil || len(bytes.TrimSpace(b)) == 0 { + fmt.Fprintln(stderr, "controller: -credential-stdin: no token on standard input") + return 2 + } + cred = bytes.TrimSpace(b) + } + return withRepos(c, stderr, func(ctx context.Context, r *repos.Service) error { + repo, err := r.Create(ctx, pos[0], *cloneURL, parsed, cred) + if err != nil { + return err + } + token, err := r.CreateToken(ctx, repo.Name, "created with the repository") + if err != nil { + return err + } + fmt.Fprintf(stdout, "Repository %s registered.\n\nRoutes (first match wins):\n", repo.Name) + for _, rt := range repo.Routes { + fmt.Fprintf(stdout, " %s\n", rt) + } + fmt.Fprintf(stdout, "\nTrigger token (shown once; store it as OPIFICA_TRIGGER_TOKEN):\n\n %s\n", token) + return nil + }) +} + +func repoList(args []string, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("repo list", flag.ContinueOnError) + fs.SetOutput(stderr) + var c common + c.register(fs) + if err := fs.Parse(args); err != nil { + return 2 + } + return withRepos(c, stderr, func(ctx context.Context, r *repos.Service) error { + list, err := r.List(ctx) + if err != nil { + return err + } + tw := tabwriter.NewWriter(stdout, 0, 4, 2, ' ', 0) + fmt.Fprintln(tw, "NAME\tCLONE URL\tCREDENTIAL\tROUTES") + for _, repo := range list { + var rs []string + for _, rt := range repo.Routes { + rs = append(rs, rt.String()) + } + cred := "no" + if repo.HasCredential { + cred = "yes" + } + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\n", repo.Name, repo.CloneURL, cred, strings.Join(rs, " ")) + } + return tw.Flush() + }) +} + +func repoDelete(args []string, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("repo delete", flag.ContinueOnError) + fs.SetOutput(stderr) + var c common + c.register(fs) + pos, err := parseWithArgs(fs, args) + if err != nil || len(pos) != 1 { + fmt.Fprintln(stderr, "Usage: controller repo delete [options] OWNER/NAME") + return 2 + } + return withRepos(c, stderr, func(ctx context.Context, r *repos.Service) error { + n, err := r.Store.DeleteRepository(ctx, pos[0]) + if err != nil { + return err + } + if n == 0 { + return fmt.Errorf("no repository named %s", pos[0]) + } + fmt.Fprintf(stdout, "Repository %s deleted; its trigger tokens no longer work. Past runs are kept.\n", pos[0]) + return nil + }) +} + +func repoTokenCreate(args []string, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("repo token create", flag.ContinueOnError) + fs.SetOutput(stderr) + var c common + c.register(fs) + desc := fs.String("description", "", "where the token is used, e.g. github-actions") + pos, err := parseWithArgs(fs, args) + if err != nil || len(pos) != 1 { + fmt.Fprintln(stderr, "Usage: controller repo token create [options] OWNER/NAME") + return 2 + } + return withRepos(c, stderr, func(ctx context.Context, r *repos.Service) error { + token, err := r.CreateToken(ctx, pos[0], *desc) + if err != nil { + return err + } + fmt.Fprintf(stdout, "Trigger token for %s (shown once):\n\n %s\n", pos[0], token) + return nil + }) +} diff --git a/cmd/opifica/client.go b/cmd/opifica/client.go new file mode 100644 index 0000000..3731c44 --- /dev/null +++ b/cmd/opifica/client.go @@ -0,0 +1,62 @@ +package main + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/url" + "os" + "time" + + "connectrpc.com/connect" + + "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" + "github.com/opifica/opifica/pkg/worker" +) + +// controllerConfig is how the CLI reaches the controller's TriggerService. +// The token only ever comes from the environment, so it doesn't show up in +// process listings or shell history. +type controllerConfig struct { + url, fingerprint, repo string +} + +func (c *controllerConfig) fromEnv() { + c.url = os.Getenv("OPIFICA_CONTROLLER") + c.fingerprint = os.Getenv("OPIFICA_CA_FINGERPRINT") + c.repo = os.Getenv("OPIFICA_REPO") +} + +func (c *controllerConfig) client() (opificav1connect.TriggerServiceClient, error) { + token := os.Getenv("OPIFICA_TRIGGER_TOKEN") + switch { + case c.url == "": + return nil, errors.New("the controller URL is required (-controller or OPIFICA_CONTROLLER)") + case c.fingerprint == "": + return nil, errors.New("the controller's CA fingerprint is required (-ca-fingerprint or OPIFICA_CA_FINGERPRINT)") + case c.repo == "": + return nil, errors.New("the repository is required (-repo or OPIFICA_REPO)") + case token == "": + return nil, errors.New("OPIFICA_TRIGGER_TOKEN must be set (create one with: controller repo token create)") + } + u, err := url.Parse(c.url) + if err != nil || u.Scheme != "https" || u.Hostname() == "" { + return nil, fmt.Errorf("controller URL must be https://host[:port], got %q", c.url) + } + tlsCfg, err := worker.PinnedCAConfig(c.fingerprint, u.Hostname()) + if err != nil { + return nil, err + } + httpClient := &http.Client{ + Timeout: time.Minute, + Transport: &http.Transport{TLSClientConfig: tlsCfg, ForceAttemptHTTP2: true}, + } + auth := connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { + return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { + req.Header().Set("Authorization", "Bearer "+token) + return next(ctx, req) + } + }) + return opificav1connect.NewTriggerServiceClient(httpClient, c.url, connect.WithInterceptors(auth)), nil +} diff --git a/cmd/opifica/github.go b/cmd/opifica/github.go new file mode 100644 index 0000000..a5c2afb --- /dev/null +++ b/cmd/opifica/github.go @@ -0,0 +1,50 @@ +package main + +import ( + "encoding/json" + "fmt" + "os" + "strings" +) + +// triggerEvent is what to report to the controller. +type triggerEvent struct { + event, ref, sha, base string +} + +// githubEvent reads the event of the GitHub Actions run we are in. +func githubEvent(getenv func(string) string) (triggerEvent, error) { + name, ref, sha := getenv("GITHUB_EVENT_NAME"), getenv("GITHUB_REF"), getenv("GITHUB_SHA") + switch name { + case "push", "workflow_dispatch": + switch { + case strings.HasPrefix(ref, "refs/heads/"): + return triggerEvent{event: "push", ref: ref, sha: sha}, nil + case strings.HasPrefix(ref, "refs/tags/"): + return triggerEvent{event: "tag", ref: ref, sha: sha}, nil + } + return triggerEvent{}, fmt.Errorf("GITHUB_REF %q is neither a branch nor a tag", ref) + case "pull_request", "pull_request_target": + // GITHUB_SHA is GitHub's temporary merge commit; build the head + // commit the author pushed instead. + data, err := os.ReadFile(getenv("GITHUB_EVENT_PATH")) + if err != nil { + return triggerEvent{}, fmt.Errorf("reading the pull request event: %w", err) + } + var payload struct { + PullRequest struct { + Head struct{ Ref, Sha string } + Base struct{ Ref string } + } `json:"pull_request"` + } + if err := json.Unmarshal(data, &payload); err != nil { + return triggerEvent{}, fmt.Errorf("reading the pull request event: %w", err) + } + pr := payload.PullRequest + if pr.Head.Sha == "" || pr.Base.Ref == "" { + return triggerEvent{}, fmt.Errorf("the pull request event has no head commit or base branch") + } + return triggerEvent{event: "pull_request", ref: "refs/heads/" + pr.Head.Ref, sha: pr.Head.Sha, base: pr.Base.Ref}, nil + } + return triggerEvent{}, fmt.Errorf("GitHub event %q is not supported (push, workflow_dispatch, pull_request)", name) +} diff --git a/cmd/opifica/github_test.go b/cmd/opifica/github_test.go new file mode 100644 index 0000000..855b669 --- /dev/null +++ b/cmd/opifica/github_test.go @@ -0,0 +1,36 @@ +package main + +import ( + "os" + "path/filepath" + "testing" +) + +func env(m map[string]string) func(string) string { return func(k string) string { return m[k] } } + +func TestGitHubEvent(t *testing.T) { + sha := "1111111111111111111111111111111111111111" + ev, err := githubEvent(env(map[string]string{"GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/heads/main", "GITHUB_SHA": sha})) + if err != nil || ev != (triggerEvent{event: "push", ref: "refs/heads/main", sha: sha}) { + t.Errorf("push: %+v, %v", ev, err) + } + ev, err = githubEvent(env(map[string]string{"GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/tags/v1.0", "GITHUB_SHA": sha})) + if err != nil || ev.event != "tag" { + t.Errorf("tag push: %+v, %v", ev, err) + } + + payload := filepath.Join(t.TempDir(), "event.json") + os.WriteFile(payload, []byte(`{"pull_request":{"head":{"ref":"fix","sha":"2222222222222222222222222222222222222222"},"base":{"ref":"main"}}}`), 0o644) + ev, err = githubEvent(env(map[string]string{ + "GITHUB_EVENT_NAME": "pull_request", "GITHUB_EVENT_PATH": payload, + "GITHUB_REF": "refs/pull/7/merge", "GITHUB_SHA": "3333333333333333333333333333333333333333", + })) + want := triggerEvent{event: "pull_request", ref: "refs/heads/fix", sha: "2222222222222222222222222222222222222222", base: "main"} + if err != nil || ev != want { + t.Errorf("pull request: %+v, %v; want the head commit, not GitHub's merge commit", ev, err) + } + + if _, err := githubEvent(env(map[string]string{"GITHUB_EVENT_NAME": "schedule"})); err == nil { + t.Error("unsupported event accepted") + } +} diff --git a/cmd/opifica/hook.go b/cmd/opifica/hook.go new file mode 100644 index 0000000..e0a7fa0 --- /dev/null +++ b/cmd/opifica/hook.go @@ -0,0 +1,73 @@ +package main + +import ( + "bufio" + "context" + "fmt" + "io" + "strings" + "time" + + "connectrpc.com/connect" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" +) + +const zeroSHA = "0000000000000000000000000000000000000000" + +// hook implements `opifica hook post-receive`: git runs it after a push with +// one "old new ref" line per updated ref on stdin. Configuration comes from +// the environment (OPIFICA_CONTROLLER, OPIFICA_CA_FINGERPRINT, OPIFICA_REPO, +// OPIFICA_TRIGGER_TOKEN), set by the hook script. +// +// It never fails: the push has already been accepted, and a broken CI must +// not look like a broken push. Problems are printed for the pusher. +func hook(args []string, stdin io.Reader, stdout io.Writer) int { + if len(args) != 1 || args[0] != "post-receive" { + fmt.Fprintln(stdout, "opifica: usage: opifica hook post-receive") + return 0 + } + var c controllerConfig + c.fromEnv() + client, err := c.client() + if err != nil { + fmt.Fprintf(stdout, "opifica: not triggering a run: %v\n", err) + return 0 + } + + sc := bufio.NewScanner(stdin) + for sc.Scan() { + fields := strings.Fields(sc.Text()) + if len(fields) != 3 { + continue + } + newSHA, ref := fields[1], fields[2] + var event string + switch { + case strings.Trim(newSHA, "0") == "": + continue // the ref was deleted + case strings.HasPrefix(ref, "refs/heads/"): + event = "push" + case strings.HasPrefix(ref, "refs/tags/"): + event = "tag" + default: + continue + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + resp, err := client.Trigger(ctx, connect.NewRequest(&opificav1.TriggerRequest{ + Repository: c.repo, Event: event, Ref: ref, CommitSha: newSHA, + })) + cancel() + name := strings.TrimPrefix(strings.TrimPrefix(ref, "refs/heads/"), "refs/tags/") + switch { + case err != nil: + fmt.Fprintf(stdout, "opifica: %s: could not trigger a run: %v\n", name, err) + case resp.Msg.GetRunId() == "": + fmt.Fprintf(stdout, "opifica: %s: %s\n", name, resp.Msg.GetMessage()) + default: + fmt.Fprintf(stdout, "opifica: %s: run %s started in pool %s (%d jobs)\n", + name, resp.Msg.GetRunId(), resp.Msg.GetPool(), resp.Msg.GetJobs()) + } + } + return 0 +} diff --git a/cmd/opifica/main.go b/cmd/opifica/main.go index 6f0619e..d4e865d 100644 --- a/cmd/opifica/main.go +++ b/cmd/opifica/main.go @@ -22,16 +22,22 @@ Commands: validate [file] Check a pipeline file (default: .opifica/pipeline.yaml) plan [file] Show which jobs an event would run schema Print the JSON Schema for pipeline files + trigger Start a run on the controller (e.g. from GitHub Actions) + hook post-receive Git hook for hosted repositories: start runs for pushes version Print the version Run "opifica -h" for a command's options. ` func main() { - os.Exit(run(os.Args[1:], os.Stdout, os.Stderr)) + os.Exit(runWithInput(os.Args[1:], os.Stdin, os.Stdout, os.Stderr)) } func run(args []string, stdout, stderr io.Writer) int { + return runWithInput(args, strings.NewReader(""), stdout, stderr) +} + +func runWithInput(args []string, stdin io.Reader, stdout, stderr io.Writer) int { if len(args) == 0 { fmt.Fprint(stderr, usage) return 2 @@ -41,6 +47,10 @@ func run(args []string, stdout, stderr io.Writer) int { return validate(args[1:], stdout, stderr) case "plan": return plan(args[1:], stdout, stderr) + case "trigger": + return trigger(args[1:], stdout, stderr) + case "hook": + return hook(args[1:], stdin, stdout) case "schema": stdout.Write(schema.Pipeline) return 0 diff --git a/cmd/opifica/trigger.go b/cmd/opifica/trigger.go new file mode 100644 index 0000000..2d97945 --- /dev/null +++ b/cmd/opifica/trigger.go @@ -0,0 +1,216 @@ +package main + +import ( + "context" + "flag" + "fmt" + "io" + "os" + "os/exec" + "strings" + "time" + + "connectrpc.com/connect" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" +) + +// Exit codes of `opifica trigger`. +const ( + exitRunFailed = 1 + exitUsage = 2 + exitError = 3 +) + +func trigger(args []string, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("trigger", flag.ContinueOnError) + fs.SetOutput(stderr) + var c controllerConfig + c.fromEnv() + fs.StringVar(&c.url, "controller", c.url, "controller URL (OPIFICA_CONTROLLER)") + fs.StringVar(&c.fingerprint, "ca-fingerprint", c.fingerprint, "controller CA fingerprint (OPIFICA_CA_FINGERPRINT)") + fs.StringVar(&c.repo, "repo", c.repo, "registered repository, e.g. acme/app (OPIFICA_REPO)") + var ev triggerEvent + fs.StringVar(&ev.event, "event", "", "push, tag or pull_request (default: from GitHub Actions, or push of the current branch)") + fs.StringVar(&ev.ref, "ref", "", "full ref, e.g. refs/heads/main") + fs.StringVar(&ev.sha, "sha", "", "commit to build") + fs.StringVar(&ev.base, "base", "", "pull request target branch") + wait := fs.Bool("wait", false, "wait for the run, print its output, and exit non-zero if it fails") + timeout := fs.Duration("timeout", 6*time.Hour, "with -wait: give up after this long") + fs.Usage = func() { + fmt.Fprint(stderr, `Usage: opifica trigger [options] + +Starts a run for a commit of a registered repository. In GitHub Actions the +event, ref and commit are read from the workflow's environment; elsewhere +they default to the current branch and commit. The trigger token is read +from OPIFICA_TRIGGER_TOKEN. + +Options: +`) + fs.PrintDefaults() + } + if err := fs.Parse(args); err != nil { + return exitUsage + } + if ev.event == "" && ev.sha == "" { + var err error + if os.Getenv("GITHUB_ACTIONS") == "true" { + ev, err = githubEvent(os.Getenv) + } else { + ev, err = localEvent() + } + if err != nil { + fmt.Fprintf(stderr, "opifica: %v\n", err) + return exitUsage + } + } + client, err := c.client() + if err != nil { + fmt.Fprintf(stderr, "opifica: %v\n", err) + return exitUsage + } + + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + resp, err := client.Trigger(ctx, connect.NewRequest(&opificav1.TriggerRequest{ + Repository: c.repo, Event: ev.event, Ref: ev.ref, CommitSha: ev.sha, BaseBranch: ev.base, + })) + if err != nil { + fmt.Fprintf(stderr, "opifica: %v\n", err) + return exitError + } + if resp.Msg.GetRunId() == "" { + fmt.Fprintf(stdout, "No run started: %s\n", resp.Msg.GetMessage()) + return 0 + } + fmt.Fprintf(stdout, "Run %s started in pool %s with %d job(s).\n", resp.Msg.GetRunId(), resp.Msg.GetPool(), resp.Msg.GetJobs()) + if !*wait { + return 0 + } + return follow(ctx, client, resp.Msg.GetRunId(), stdout, stderr) +} + +// localEvent is a push of the current branch at HEAD. +func localEvent() (triggerEvent, error) { + sha, err := gitOutput("rev-parse", "HEAD") + if err != nil { + return triggerEvent{}, fmt.Errorf("finding the current commit: %w", err) + } + branch, err := gitOutput("symbolic-ref", "--short", "HEAD") + if err != nil { + return triggerEvent{}, fmt.Errorf("not on a branch; pass -event, -ref and -sha") + } + return triggerEvent{event: "push", ref: "refs/heads/" + branch, sha: sha}, nil +} + +func gitOutput(args ...string) (string, error) { + out, err := exec.Command("git", args...).Output() + return strings.TrimSpace(string(out)), err +} + +// follow prints the run's output as it arrives and returns once it ends. +func follow(ctx context.Context, client opificav1connect.TriggerServiceClient, runID string, stdout, stderr io.Writer) int { + offsets := map[string]int64{} + done := map[string]bool{} + var last *opificav1.GetRunResponse + for { + run, err := client.GetRun(ctx, connect.NewRequest(&opificav1.GetRunRequest{RunId: runID})) + if err != nil { + fmt.Fprintf(stderr, "opifica: %v\n", err) + return exitError + } + last = run.Msg + for _, j := range last.GetJobs() { + if done[j.GetId()] { + continue + } + for { + lg, err := client.GetJobLog(ctx, connect.NewRequest(&opificav1.GetJobLogRequest{JobId: j.GetId(), Offset: offsets[j.GetId()]})) + if err != nil { + fmt.Fprintf(stderr, "opifica: %v\n", err) + return exitError + } + printPrefixed(stdout, j.GetName(), lg.Msg.GetData()) + offsets[j.GetId()] = lg.Msg.GetNextOffset() + if lg.Msg.GetComplete() { + done[j.GetId()] = true + } + if len(lg.Msg.GetData()) == 0 || lg.Msg.GetComplete() { + break + } + } + } + if last.GetState() != "running" && allDone(last, done) { + break + } + select { + case <-time.After(2 * time.Second): + case <-ctx.Done(): + fmt.Fprintln(stderr, "opifica: gave up waiting for the run") + return exitError + } + } + + fmt.Fprintf(stdout, "\nRun %s: %s\n", runID, last.GetState()) + for _, j := range last.GetJobs() { + line := fmt.Sprintf(" %-10s %s", j.GetState(), j.GetName()) + if j.ExitCode != nil && j.GetExitCode() != 0 { + line += fmt.Sprintf(" (exit %d)", j.GetExitCode()) + } + if j.GetError() != "" { + line += ": " + j.GetError() + } + fmt.Fprintln(stdout, line) + } + writeStepSummary(runID, last) + if last.GetState() != "succeeded" { + return exitRunFailed + } + return 0 +} + +func allDone(run *opificav1.GetRunResponse, done map[string]bool) bool { + for _, j := range run.GetJobs() { + if !done[j.GetId()] { + return false + } + } + return true +} + +// printPrefixed prints log data with the job's name before each line. +func printPrefixed(w io.Writer, job string, data []byte) { + if len(data) == 0 { + return + } + for _, line := range strings.SplitAfter(string(data), "\n") { + if line != "" { + fmt.Fprintf(w, "[%s] %s", job, line) + } + } + if data[len(data)-1] != '\n' { + fmt.Fprintln(w) + } +} + +// writeStepSummary adds a results table to the GitHub Actions run page. +func writeStepSummary(runID string, run *opificav1.GetRunResponse) { + path := os.Getenv("GITHUB_STEP_SUMMARY") + if path == "" { + return + } + f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0o644) + if err != nil { + return + } + defer f.Close() + fmt.Fprintf(f, "### Opifica run `%s`: %s\n\n| Job | OS | Result |\n| --- | --- | --- |\n", runID, run.GetState()) + for _, j := range run.GetJobs() { + result := j.GetState() + if j.GetError() != "" { + result += ": " + j.GetError() + } + fmt.Fprintf(f, "| %s | %s | %s |\n", j.GetName(), j.GetOs(), strings.ReplaceAll(result, "|", "\\|")) + } +} diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 29183ec..145d8fb 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -3,9 +3,9 @@ Where Opifica is and what comes next. Each milestone ends in something that can be run and tested. Tick items off as they land. -**Defaults chosen for now** (can be revisited): -first Git provider is **Forgejo/Gitea**, **Linux** VMs come before Windows and macOS, and logs are stored on the -**controller's local disk** until S3 storage is added. +**Decisions so far:** repositories live on **your own git server** (bare repositories managed by Opifica) or on +**GitHub** (runs triggered from a GitHub Actions step); there are no forge integrations. **Linux** VMs come before +Windows and macOS, and logs are stored on the **controller's disk** until S3 storage is added. ## M0 · Foundations ✅ @@ -63,16 +63,22 @@ result are stored. Tested in ordinary CI, using the fake hypervisor with a real - [x] Integration test: controller + PostgreSQL + worker (fake hypervisor) + real agent, in CI - [x] Failure tests: worker killed mid-job, lease expiry, reconnect, denied secret -## M2 · Triggered by Git +## M2 · Triggered by git: your own server or GitHub -**Goal:** a push to a Forgejo repository starts a run, and the commit shows its status. +**Goal:** a `git push` starts a run, whether the repository is hosted by Opifica or on GitHub. -- [ ] Provider interface and Forgejo/Gitea provider: webhooks with signature checks, `FetchFile` at a commit, - short-lived clone credentials, commit status -- [ ] Routing events to pools from the admin config (pull requests → `untrusted`, `main` → `trusted`, - tags → `release`) -- [ ] Integration test against a Forgejo container -- [ ] `docker compose` dev environment: PostgreSQL, Forgejo, controller, worker +- [x] Repository registry: clone URL, routing rules (event + ref → pool, first match wins), trigger tokens + (`controller repo create|list|delete|token create`) +- [x] Trigger API; the controller reads the pipeline file at the commit itself and picks the pool from the + repository's rules +- [x] `opifica hook post-receive`: the git hook for bare repositories; prints the run for the pusher +- [x] `opifica trigger -wait` for GitHub Actions: passes or fails with the run, prints its output, writes a + step summary ([example workflow](../examples/github-actions.yml)) +- [x] Optional read credential per repository, encrypted, given to workers with each job +- [ ] **opifica-git**: create a repository through the API or CLI, which creates the bare repository and installs + the hook; serves git to developers (HTTPS and/or SSH: to decide); users and access tokens +- [ ] Workers fetch hosted repositories with short-lived, read-only tokens +- [ ] `docker compose` dev environment: PostgreSQL, controller, opifica-git, devworker ## M3 · Real Linux VMs @@ -113,7 +119,6 @@ Needs self-hosted hardware: an Apple silicon Mac and a Windows machine with Hype ## M7 · Everything else - [ ] Web UI (Angular): runs, live logs, test results, workers, administration -- [ ] GitHub and GitLab providers; plain Git with polling - [ ] GitOps sync of the admin config from a Git repository - [ ] S3-compatible storage for logs and artifacts - [ ] OIDC login; OpenBao/Vault as a secrets backend diff --git a/examples/github-actions.yml b/examples/github-actions.yml new file mode 100644 index 0000000..f7ab8fd --- /dev/null +++ b/examples/github-actions.yml @@ -0,0 +1,36 @@ +# Runs Opifica for a repository hosted on GitHub. +# +# Copy to .github/workflows/opifica.yml. In the repository's settings, add: +# variables: OPIFICA_CONTROLLER (https://your-controller:8443) +# OPIFICA_CA_FINGERPRINT (sha256:..., from `controller token create`) +# secret: OPIFICA_TRIGGER_TOKEN (from `controller repo create` or `repo token create`) +# +# The controller must be reachable from the runner: run this on a self-hosted +# runner inside your network, or expose the controller's port. +# +# The step passes or fails with the Opifica run, prints its output, and adds +# a summary table to the workflow run. +name: Opifica + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + opifica: + runs-on: ubuntu-latest + steps: + - uses: actions/setup-go@v7 + with: + go-version: "1.27" + # Until signed release binaries exist (see docs/ROADMAP.md, M4). + - run: go install github.com/opifica/opifica/cmd/opifica@main + - run: opifica trigger -wait + env: + OPIFICA_CONTROLLER: ${{ vars.OPIFICA_CONTROLLER }} + OPIFICA_CA_FINGERPRINT: ${{ vars.OPIFICA_CA_FINGERPRINT }} + OPIFICA_REPO: ${{ github.repository }} + OPIFICA_TRIGGER_TOKEN: ${{ secrets.OPIFICA_TRIGGER_TOKEN }} diff --git a/gen/opifica/v1/opificav1connect/trigger.connect.go b/gen/opifica/v1/opificav1connect/trigger.connect.go new file mode 100644 index 0000000..dccd998 --- /dev/null +++ b/gen/opifica/v1/opificav1connect/trigger.connect.go @@ -0,0 +1,175 @@ +// Code generated by protoc-gen-connect-go. DO NOT EDIT. +// +// Source: opifica/v1/trigger.proto + +package opificav1connect + +import ( + connect "connectrpc.com/connect" + context "context" + errors "errors" + v1 "github.com/opifica/opifica/gen/opifica/v1" + http "net/http" + strings "strings" +) + +// This is a compile-time assertion to ensure that this generated file and the connect package are +// compatible. If you get a compiler error that this constant is not defined, this code was +// generated with a version of connect newer than the one compiled into your binary. You can fix the +// problem by either regenerating this code with an older version of connect or updating the connect +// version compiled into your binary. +const _ = connect.IsAtLeastVersion1_13_0 + +const ( + // TriggerServiceName is the fully-qualified name of the TriggerService service. + TriggerServiceName = "opifica.v1.TriggerService" +) + +// These constants are the fully-qualified names of the RPCs defined in this package. They're +// exposed at runtime as Spec.Procedure and as the final two segments of the HTTP route. +// +// Note that these are different from the fully-qualified method names used by +// google.golang.org/protobuf/reflect/protoreflect. To convert from these constants to +// reflection-formatted method names, remove the leading slash and convert the remaining slash to a +// period. +const ( + // TriggerServiceTriggerProcedure is the fully-qualified name of the TriggerService's Trigger RPC. + TriggerServiceTriggerProcedure = "/opifica.v1.TriggerService/Trigger" + // TriggerServiceGetRunProcedure is the fully-qualified name of the TriggerService's GetRun RPC. + TriggerServiceGetRunProcedure = "/opifica.v1.TriggerService/GetRun" + // TriggerServiceGetJobLogProcedure is the fully-qualified name of the TriggerService's GetJobLog + // RPC. + TriggerServiceGetJobLogProcedure = "/opifica.v1.TriggerService/GetJobLog" +) + +// TriggerServiceClient is a client for the opifica.v1.TriggerService service. +type TriggerServiceClient interface { + // Trigger reports that a ref moved. The controller reads the pipeline file + // at the commit, routes the run to a pool with the repository's rules and + // schedules it. + Trigger(context.Context, *connect.Request[v1.TriggerRequest]) (*connect.Response[v1.TriggerResponse], error) + // GetRun returns a run's state and jobs, e.g. for `opifica trigger --wait`. + GetRun(context.Context, *connect.Request[v1.GetRunRequest]) (*connect.Response[v1.GetRunResponse], error) + // GetJobLog returns a job's output from an offset, so callers can follow it. + GetJobLog(context.Context, *connect.Request[v1.GetJobLogRequest]) (*connect.Response[v1.GetJobLogResponse], error) +} + +// NewTriggerServiceClient constructs a client for the opifica.v1.TriggerService service. By +// default, it uses the Connect protocol with the binary Protobuf Codec, asks for gzipped responses, +// and sends uncompressed requests. To use the gRPC or gRPC-Web protocols, supply the +// connect.WithGRPC() or connect.WithGRPCWeb() options. +// +// The URL supplied here should be the base URL for the Connect or gRPC server (for example, +// http://api.acme.com or https://acme.com/grpc). +func NewTriggerServiceClient(httpClient connect.HTTPClient, baseURL string, opts ...connect.ClientOption) TriggerServiceClient { + baseURL = strings.TrimRight(baseURL, "/") + triggerServiceMethods := v1.File_opifica_v1_trigger_proto.Services().ByName("TriggerService").Methods() + return &triggerServiceClient{ + trigger: connect.NewClient[v1.TriggerRequest, v1.TriggerResponse]( + httpClient, + baseURL+TriggerServiceTriggerProcedure, + connect.WithSchema(triggerServiceMethods.ByName("Trigger")), + connect.WithClientOptions(opts...), + ), + getRun: connect.NewClient[v1.GetRunRequest, v1.GetRunResponse]( + httpClient, + baseURL+TriggerServiceGetRunProcedure, + connect.WithSchema(triggerServiceMethods.ByName("GetRun")), + connect.WithClientOptions(opts...), + ), + getJobLog: connect.NewClient[v1.GetJobLogRequest, v1.GetJobLogResponse]( + httpClient, + baseURL+TriggerServiceGetJobLogProcedure, + connect.WithSchema(triggerServiceMethods.ByName("GetJobLog")), + connect.WithClientOptions(opts...), + ), + } +} + +// triggerServiceClient implements TriggerServiceClient. +type triggerServiceClient struct { + trigger *connect.Client[v1.TriggerRequest, v1.TriggerResponse] + getRun *connect.Client[v1.GetRunRequest, v1.GetRunResponse] + getJobLog *connect.Client[v1.GetJobLogRequest, v1.GetJobLogResponse] +} + +// Trigger calls opifica.v1.TriggerService.Trigger. +func (c *triggerServiceClient) Trigger(ctx context.Context, req *connect.Request[v1.TriggerRequest]) (*connect.Response[v1.TriggerResponse], error) { + return c.trigger.CallUnary(ctx, req) +} + +// GetRun calls opifica.v1.TriggerService.GetRun. +func (c *triggerServiceClient) GetRun(ctx context.Context, req *connect.Request[v1.GetRunRequest]) (*connect.Response[v1.GetRunResponse], error) { + return c.getRun.CallUnary(ctx, req) +} + +// GetJobLog calls opifica.v1.TriggerService.GetJobLog. +func (c *triggerServiceClient) GetJobLog(ctx context.Context, req *connect.Request[v1.GetJobLogRequest]) (*connect.Response[v1.GetJobLogResponse], error) { + return c.getJobLog.CallUnary(ctx, req) +} + +// TriggerServiceHandler is an implementation of the opifica.v1.TriggerService service. +type TriggerServiceHandler interface { + // Trigger reports that a ref moved. The controller reads the pipeline file + // at the commit, routes the run to a pool with the repository's rules and + // schedules it. + Trigger(context.Context, *connect.Request[v1.TriggerRequest]) (*connect.Response[v1.TriggerResponse], error) + // GetRun returns a run's state and jobs, e.g. for `opifica trigger --wait`. + GetRun(context.Context, *connect.Request[v1.GetRunRequest]) (*connect.Response[v1.GetRunResponse], error) + // GetJobLog returns a job's output from an offset, so callers can follow it. + GetJobLog(context.Context, *connect.Request[v1.GetJobLogRequest]) (*connect.Response[v1.GetJobLogResponse], error) +} + +// NewTriggerServiceHandler builds an HTTP handler from the service implementation. It returns the +// path on which to mount the handler and the handler itself. +// +// By default, handlers support the Connect, gRPC, and gRPC-Web protocols with the binary Protobuf +// and JSON codecs. They also support gzip compression. +func NewTriggerServiceHandler(svc TriggerServiceHandler, opts ...connect.HandlerOption) (string, http.Handler) { + triggerServiceMethods := v1.File_opifica_v1_trigger_proto.Services().ByName("TriggerService").Methods() + triggerServiceTriggerHandler := connect.NewUnaryHandler( + TriggerServiceTriggerProcedure, + svc.Trigger, + connect.WithSchema(triggerServiceMethods.ByName("Trigger")), + connect.WithHandlerOptions(opts...), + ) + triggerServiceGetRunHandler := connect.NewUnaryHandler( + TriggerServiceGetRunProcedure, + svc.GetRun, + connect.WithSchema(triggerServiceMethods.ByName("GetRun")), + connect.WithHandlerOptions(opts...), + ) + triggerServiceGetJobLogHandler := connect.NewUnaryHandler( + TriggerServiceGetJobLogProcedure, + svc.GetJobLog, + connect.WithSchema(triggerServiceMethods.ByName("GetJobLog")), + connect.WithHandlerOptions(opts...), + ) + return "/opifica.v1.TriggerService/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case TriggerServiceTriggerProcedure: + triggerServiceTriggerHandler.ServeHTTP(w, r) + case TriggerServiceGetRunProcedure: + triggerServiceGetRunHandler.ServeHTTP(w, r) + case TriggerServiceGetJobLogProcedure: + triggerServiceGetJobLogHandler.ServeHTTP(w, r) + default: + http.NotFound(w, r) + } + }) +} + +// UnimplementedTriggerServiceHandler returns CodeUnimplemented from all methods. +type UnimplementedTriggerServiceHandler struct{} + +func (UnimplementedTriggerServiceHandler) Trigger(context.Context, *connect.Request[v1.TriggerRequest]) (*connect.Response[v1.TriggerResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("opifica.v1.TriggerService.Trigger is not implemented")) +} + +func (UnimplementedTriggerServiceHandler) GetRun(context.Context, *connect.Request[v1.GetRunRequest]) (*connect.Response[v1.GetRunResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("opifica.v1.TriggerService.GetRun is not implemented")) +} + +func (UnimplementedTriggerServiceHandler) GetJobLog(context.Context, *connect.Request[v1.GetJobLogRequest]) (*connect.Response[v1.GetJobLogResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("opifica.v1.TriggerService.GetJobLog is not implemented")) +} diff --git a/gen/opifica/v1/trigger.pb.go b/gen/opifica/v1/trigger.pb.go new file mode 100644 index 0000000..8cd3b9b --- /dev/null +++ b/gen/opifica/v1/trigger.pb.go @@ -0,0 +1,580 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.12 +// protoc (unknown) +// source: opifica/v1/trigger.proto + +package opificav1 + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type TriggerRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Registered repository name, e.g. "acme/app". + Repository string `protobuf:"bytes,1,opt,name=repository,proto3" json:"repository,omitempty"` + // "push", "tag" or "pull_request". + Event string `protobuf:"bytes,2,opt,name=event,proto3" json:"event,omitempty"` + // Full ref: refs/heads/main, refs/tags/v1.0. For pull requests, the ref of + // the head branch. + Ref string `protobuf:"bytes,3,opt,name=ref,proto3" json:"ref,omitempty"` + // The exact commit to build. + CommitSha string `protobuf:"bytes,4,opt,name=commit_sha,json=commitSha,proto3" json:"commit_sha,omitempty"` + // Pull requests only: the branch the pull request targets. + BaseBranch string `protobuf:"bytes,5,opt,name=base_branch,json=baseBranch,proto3" json:"base_branch,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TriggerRequest) Reset() { + *x = TriggerRequest{} + mi := &file_opifica_v1_trigger_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TriggerRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TriggerRequest) ProtoMessage() {} + +func (x *TriggerRequest) ProtoReflect() protoreflect.Message { + mi := &file_opifica_v1_trigger_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TriggerRequest.ProtoReflect.Descriptor instead. +func (*TriggerRequest) Descriptor() ([]byte, []int) { + return file_opifica_v1_trigger_proto_rawDescGZIP(), []int{0} +} + +func (x *TriggerRequest) GetRepository() string { + if x != nil { + return x.Repository + } + return "" +} + +func (x *TriggerRequest) GetEvent() string { + if x != nil { + return x.Event + } + return "" +} + +func (x *TriggerRequest) GetRef() string { + if x != nil { + return x.Ref + } + return "" +} + +func (x *TriggerRequest) GetCommitSha() string { + if x != nil { + return x.CommitSha + } + return "" +} + +func (x *TriggerRequest) GetBaseBranch() string { + if x != nil { + return x.BaseBranch + } + return "" +} + +type TriggerResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Empty when no run was started; message says why. + RunId string `protobuf:"bytes,1,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"` + Pool string `protobuf:"bytes,2,opt,name=pool,proto3" json:"pool,omitempty"` + Jobs int32 `protobuf:"varint,3,opt,name=jobs,proto3" json:"jobs,omitempty"` + Message string `protobuf:"bytes,4,opt,name=message,proto3" json:"message,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TriggerResponse) Reset() { + *x = TriggerResponse{} + mi := &file_opifica_v1_trigger_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TriggerResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TriggerResponse) ProtoMessage() {} + +func (x *TriggerResponse) ProtoReflect() protoreflect.Message { + mi := &file_opifica_v1_trigger_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TriggerResponse.ProtoReflect.Descriptor instead. +func (*TriggerResponse) Descriptor() ([]byte, []int) { + return file_opifica_v1_trigger_proto_rawDescGZIP(), []int{1} +} + +func (x *TriggerResponse) GetRunId() string { + if x != nil { + return x.RunId + } + return "" +} + +func (x *TriggerResponse) GetPool() string { + if x != nil { + return x.Pool + } + return "" +} + +func (x *TriggerResponse) GetJobs() int32 { + if x != nil { + return x.Jobs + } + return 0 +} + +func (x *TriggerResponse) GetMessage() string { + if x != nil { + return x.Message + } + return "" +} + +type GetRunRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + RunId string `protobuf:"bytes,1,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetRunRequest) Reset() { + *x = GetRunRequest{} + mi := &file_opifica_v1_trigger_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetRunRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetRunRequest) ProtoMessage() {} + +func (x *GetRunRequest) ProtoReflect() protoreflect.Message { + mi := &file_opifica_v1_trigger_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetRunRequest.ProtoReflect.Descriptor instead. +func (*GetRunRequest) Descriptor() ([]byte, []int) { + return file_opifica_v1_trigger_proto_rawDescGZIP(), []int{2} +} + +func (x *GetRunRequest) GetRunId() string { + if x != nil { + return x.RunId + } + return "" +} + +type GetRunResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // "running", "succeeded", "failed" or "cancelled". + State string `protobuf:"bytes,1,opt,name=state,proto3" json:"state,omitempty"` + Jobs []*JobSummary `protobuf:"bytes,2,rep,name=jobs,proto3" json:"jobs,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetRunResponse) Reset() { + *x = GetRunResponse{} + mi := &file_opifica_v1_trigger_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetRunResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetRunResponse) ProtoMessage() {} + +func (x *GetRunResponse) ProtoReflect() protoreflect.Message { + mi := &file_opifica_v1_trigger_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetRunResponse.ProtoReflect.Descriptor instead. +func (*GetRunResponse) Descriptor() ([]byte, []int) { + return file_opifica_v1_trigger_proto_rawDescGZIP(), []int{3} +} + +func (x *GetRunResponse) GetState() string { + if x != nil { + return x.State + } + return "" +} + +func (x *GetRunResponse) GetJobs() []*JobSummary { + if x != nil { + return x.Jobs + } + return nil +} + +type JobSummary struct { + state protoimpl.MessageState `protogen:"open.v1"` + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` + Os string `protobuf:"bytes,3,opt,name=os,proto3" json:"os,omitempty"` + State string `protobuf:"bytes,4,opt,name=state,proto3" json:"state,omitempty"` + ExitCode *int32 `protobuf:"varint,5,opt,name=exit_code,json=exitCode,proto3,oneof" json:"exit_code,omitempty"` + Error string `protobuf:"bytes,6,opt,name=error,proto3" json:"error,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *JobSummary) Reset() { + *x = JobSummary{} + mi := &file_opifica_v1_trigger_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *JobSummary) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*JobSummary) ProtoMessage() {} + +func (x *JobSummary) ProtoReflect() protoreflect.Message { + mi := &file_opifica_v1_trigger_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use JobSummary.ProtoReflect.Descriptor instead. +func (*JobSummary) Descriptor() ([]byte, []int) { + return file_opifica_v1_trigger_proto_rawDescGZIP(), []int{4} +} + +func (x *JobSummary) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *JobSummary) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *JobSummary) GetOs() string { + if x != nil { + return x.Os + } + return "" +} + +func (x *JobSummary) GetState() string { + if x != nil { + return x.State + } + return "" +} + +func (x *JobSummary) GetExitCode() int32 { + if x != nil && x.ExitCode != nil { + return *x.ExitCode + } + return 0 +} + +func (x *JobSummary) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +type GetJobLogRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + JobId string `protobuf:"bytes,1,opt,name=job_id,json=jobId,proto3" json:"job_id,omitempty"` + // Byte offset to read from. + Offset int64 `protobuf:"varint,2,opt,name=offset,proto3" json:"offset,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetJobLogRequest) Reset() { + *x = GetJobLogRequest{} + mi := &file_opifica_v1_trigger_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetJobLogRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetJobLogRequest) ProtoMessage() {} + +func (x *GetJobLogRequest) ProtoReflect() protoreflect.Message { + mi := &file_opifica_v1_trigger_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetJobLogRequest.ProtoReflect.Descriptor instead. +func (*GetJobLogRequest) Descriptor() ([]byte, []int) { + return file_opifica_v1_trigger_proto_rawDescGZIP(), []int{5} +} + +func (x *GetJobLogRequest) GetJobId() string { + if x != nil { + return x.JobId + } + return "" +} + +func (x *GetJobLogRequest) GetOffset() int64 { + if x != nil { + return x.Offset + } + return 0 +} + +type GetJobLogResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` + // Offset to ask for next time. + NextOffset int64 `protobuf:"varint,2,opt,name=next_offset,json=nextOffset,proto3" json:"next_offset,omitempty"` + // True once the job has finished and data reaches the end of its log. + Complete bool `protobuf:"varint,3,opt,name=complete,proto3" json:"complete,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetJobLogResponse) Reset() { + *x = GetJobLogResponse{} + mi := &file_opifica_v1_trigger_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetJobLogResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetJobLogResponse) ProtoMessage() {} + +func (x *GetJobLogResponse) ProtoReflect() protoreflect.Message { + mi := &file_opifica_v1_trigger_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetJobLogResponse.ProtoReflect.Descriptor instead. +func (*GetJobLogResponse) Descriptor() ([]byte, []int) { + return file_opifica_v1_trigger_proto_rawDescGZIP(), []int{6} +} + +func (x *GetJobLogResponse) GetData() []byte { + if x != nil { + return x.Data + } + return nil +} + +func (x *GetJobLogResponse) GetNextOffset() int64 { + if x != nil { + return x.NextOffset + } + return 0 +} + +func (x *GetJobLogResponse) GetComplete() bool { + if x != nil { + return x.Complete + } + return false +} + +var File_opifica_v1_trigger_proto protoreflect.FileDescriptor + +const file_opifica_v1_trigger_proto_rawDesc = "" + + "\n" + + "\x18opifica/v1/trigger.proto\x12\n" + + "opifica.v1\"\x98\x01\n" + + "\x0eTriggerRequest\x12\x1e\n" + + "\n" + + "repository\x18\x01 \x01(\tR\n" + + "repository\x12\x14\n" + + "\x05event\x18\x02 \x01(\tR\x05event\x12\x10\n" + + "\x03ref\x18\x03 \x01(\tR\x03ref\x12\x1d\n" + + "\n" + + "commit_sha\x18\x04 \x01(\tR\tcommitSha\x12\x1f\n" + + "\vbase_branch\x18\x05 \x01(\tR\n" + + "baseBranch\"j\n" + + "\x0fTriggerResponse\x12\x15\n" + + "\x06run_id\x18\x01 \x01(\tR\x05runId\x12\x12\n" + + "\x04pool\x18\x02 \x01(\tR\x04pool\x12\x12\n" + + "\x04jobs\x18\x03 \x01(\x05R\x04jobs\x12\x18\n" + + "\amessage\x18\x04 \x01(\tR\amessage\"&\n" + + "\rGetRunRequest\x12\x15\n" + + "\x06run_id\x18\x01 \x01(\tR\x05runId\"R\n" + + "\x0eGetRunResponse\x12\x14\n" + + "\x05state\x18\x01 \x01(\tR\x05state\x12*\n" + + "\x04jobs\x18\x02 \x03(\v2\x16.opifica.v1.JobSummaryR\x04jobs\"\x9c\x01\n" + + "\n" + + "JobSummary\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id\x12\x12\n" + + "\x04name\x18\x02 \x01(\tR\x04name\x12\x0e\n" + + "\x02os\x18\x03 \x01(\tR\x02os\x12\x14\n" + + "\x05state\x18\x04 \x01(\tR\x05state\x12 \n" + + "\texit_code\x18\x05 \x01(\x05H\x00R\bexitCode\x88\x01\x01\x12\x14\n" + + "\x05error\x18\x06 \x01(\tR\x05errorB\f\n" + + "\n" + + "_exit_code\"A\n" + + "\x10GetJobLogRequest\x12\x15\n" + + "\x06job_id\x18\x01 \x01(\tR\x05jobId\x12\x16\n" + + "\x06offset\x18\x02 \x01(\x03R\x06offset\"d\n" + + "\x11GetJobLogResponse\x12\x12\n" + + "\x04data\x18\x01 \x01(\fR\x04data\x12\x1f\n" + + "\vnext_offset\x18\x02 \x01(\x03R\n" + + "nextOffset\x12\x1a\n" + + "\bcomplete\x18\x03 \x01(\bR\bcomplete2\xdf\x01\n" + + "\x0eTriggerService\x12B\n" + + "\aTrigger\x12\x1a.opifica.v1.TriggerRequest\x1a\x1b.opifica.v1.TriggerResponse\x12?\n" + + "\x06GetRun\x12\x19.opifica.v1.GetRunRequest\x1a\x1a.opifica.v1.GetRunResponse\x12H\n" + + "\tGetJobLog\x12\x1c.opifica.v1.GetJobLogRequest\x1a\x1d.opifica.v1.GetJobLogResponseB5Z3github.com/opifica/opifica/gen/opifica/v1;opificav1b\x06proto3" + +var ( + file_opifica_v1_trigger_proto_rawDescOnce sync.Once + file_opifica_v1_trigger_proto_rawDescData []byte +) + +func file_opifica_v1_trigger_proto_rawDescGZIP() []byte { + file_opifica_v1_trigger_proto_rawDescOnce.Do(func() { + file_opifica_v1_trigger_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_opifica_v1_trigger_proto_rawDesc), len(file_opifica_v1_trigger_proto_rawDesc))) + }) + return file_opifica_v1_trigger_proto_rawDescData +} + +var file_opifica_v1_trigger_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_opifica_v1_trigger_proto_goTypes = []any{ + (*TriggerRequest)(nil), // 0: opifica.v1.TriggerRequest + (*TriggerResponse)(nil), // 1: opifica.v1.TriggerResponse + (*GetRunRequest)(nil), // 2: opifica.v1.GetRunRequest + (*GetRunResponse)(nil), // 3: opifica.v1.GetRunResponse + (*JobSummary)(nil), // 4: opifica.v1.JobSummary + (*GetJobLogRequest)(nil), // 5: opifica.v1.GetJobLogRequest + (*GetJobLogResponse)(nil), // 6: opifica.v1.GetJobLogResponse +} +var file_opifica_v1_trigger_proto_depIdxs = []int32{ + 4, // 0: opifica.v1.GetRunResponse.jobs:type_name -> opifica.v1.JobSummary + 0, // 1: opifica.v1.TriggerService.Trigger:input_type -> opifica.v1.TriggerRequest + 2, // 2: opifica.v1.TriggerService.GetRun:input_type -> opifica.v1.GetRunRequest + 5, // 3: opifica.v1.TriggerService.GetJobLog:input_type -> opifica.v1.GetJobLogRequest + 1, // 4: opifica.v1.TriggerService.Trigger:output_type -> opifica.v1.TriggerResponse + 3, // 5: opifica.v1.TriggerService.GetRun:output_type -> opifica.v1.GetRunResponse + 6, // 6: opifica.v1.TriggerService.GetJobLog:output_type -> opifica.v1.GetJobLogResponse + 4, // [4:7] is the sub-list for method output_type + 1, // [1:4] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name +} + +func init() { file_opifica_v1_trigger_proto_init() } +func file_opifica_v1_trigger_proto_init() { + if File_opifica_v1_trigger_proto != nil { + return + } + file_opifica_v1_trigger_proto_msgTypes[4].OneofWrappers = []any{} + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_opifica_v1_trigger_proto_rawDesc), len(file_opifica_v1_trigger_proto_rawDesc)), + NumEnums: 0, + NumMessages: 7, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_opifica_v1_trigger_proto_goTypes, + DependencyIndexes: file_opifica_v1_trigger_proto_depIdxs, + MessageInfos: file_opifica_v1_trigger_proto_msgTypes, + }.Build() + File_opifica_v1_trigger_proto = out.File + file_opifica_v1_trigger_proto_goTypes = nil + file_opifica_v1_trigger_proto_depIdxs = nil +} diff --git a/internal/controller/api/api.go b/internal/controller/api/api.go index a2dba33..5903c8d 100644 --- a/internal/controller/api/api.go +++ b/internal/controller/api/api.go @@ -11,15 +11,28 @@ import ( opificav1 "github.com/opifica/opifica/gen/opifica/v1" "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" "github.com/opifica/opifica/internal/controller/ca" + "github.com/opifica/opifica/internal/controller/logs" + "github.com/opifica/opifica/internal/controller/repos" "github.com/opifica/opifica/internal/controller/workers" "github.com/opifica/opifica/internal/server" ) +// Deps are the services behind the API. +type Deps struct { + Workers *workers.Service + // Repos and Logs serve the TriggerService; without Repos it is off. + Repos *repos.Service + Logs *logs.Store +} + // NewHandler returns the controller's HTTP handler. -func NewHandler(svc *workers.Service) http.Handler { +func NewHandler(d Deps) http.Handler { mux := http.NewServeMux() mux.HandleFunc("GET /healthz", server.Health) - mux.Handle(opificav1connect.NewWorkerServiceHandler(&workerHandler{svc: svc})) + mux.Handle(opificav1connect.NewWorkerServiceHandler(&workerHandler{svc: d.Workers})) + if d.Repos != nil { + mux.Handle(opificav1connect.NewTriggerServiceHandler(&triggerHandler{repos: d.Repos, logs: d.Logs})) + } return workers.WithPeerCert(mux) } diff --git a/internal/controller/api/trigger.go b/internal/controller/api/trigger.go new file mode 100644 index 0000000..688c0ff --- /dev/null +++ b/internal/controller/api/trigger.go @@ -0,0 +1,123 @@ +package api + +import ( + "context" + "errors" + "strings" + + "connectrpc.com/connect" + "github.com/google/uuid" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" + "github.com/opifica/opifica/internal/controller/logs" + "github.com/opifica/opifica/internal/controller/repos" + "github.com/opifica/opifica/internal/controller/store" +) + +const maxLogChunk = 256 << 10 + +// triggerHandler implements TriggerService. Every call is authorized by a +// repository's trigger token and only reaches that repository's runs. +type triggerHandler struct { + opificav1connect.UnimplementedTriggerServiceHandler + repos *repos.Service + logs *logs.Store +} + +var ( + errNoToken = connect.NewError(connect.CodeUnauthenticated, errors.New("a trigger token is required (Authorization: Bearer opr_...)")) + errBadToken = connect.NewError(connect.CodeUnauthenticated, repos.ErrBadToken) + errNotYours = connect.NewError(connect.CodePermissionDenied, errors.New("this token is for another repository")) + errNoSuchRun = connect.NewError(connect.CodeNotFound, errors.New("no such run for this repository")) +) + +func (h *triggerHandler) auth(ctx context.Context, header string) (repos.Repository, error) { + token, ok := strings.CutPrefix(header, "Bearer ") + if !ok || token == "" { + return repos.Repository{}, errNoToken + } + repo, err := h.repos.Authenticate(ctx, token) + if errors.Is(err, repos.ErrBadToken) { + return repos.Repository{}, errBadToken + } + if err != nil { + return repos.Repository{}, connect.NewError(connect.CodeUnavailable, errors.New("try again later")) + } + return repo, nil +} + +func (h *triggerHandler) Trigger(ctx context.Context, req *connect.Request[opificav1.TriggerRequest]) (*connect.Response[opificav1.TriggerResponse], error) { + repo, err := h.auth(ctx, req.Header().Get("Authorization")) + if err != nil { + return nil, err + } + if req.Msg.GetRepository() != repo.Name { + return nil, errNotYours + } + out, err := h.repos.Trigger(ctx, repo, req.Msg) + if err != nil { + return nil, connect.NewError(connect.CodeFailedPrecondition, err) + } + resp := &opificav1.TriggerResponse{Pool: out.Pool, Jobs: int32(out.Jobs), Message: out.Message} + if out.RunID != uuid.Nil { + resp.RunId = out.RunID.String() + } + return connect.NewResponse(resp), nil +} + +func (h *triggerHandler) GetRun(ctx context.Context, req *connect.Request[opificav1.GetRunRequest]) (*connect.Response[opificav1.GetRunResponse], error) { + repo, err := h.auth(ctx, req.Header().Get("Authorization")) + if err != nil { + return nil, err + } + id, err := uuid.Parse(req.Msg.GetRunId()) + if err != nil { + return nil, errNoSuchRun + } + run, err := h.repos.Store.GetRun(ctx, id) + if err != nil || run.RepositoryID == nil || *run.RepositoryID != repo.ID { + return nil, errNoSuchRun + } + jobs, err := h.repos.Store.ListRunJobs(ctx, id) + if err != nil { + return nil, err + } + resp := &opificav1.GetRunResponse{State: run.State} + for _, j := range jobs { + resp.Jobs = append(resp.Jobs, &opificav1.JobSummary{ + Id: j.ID.String(), Name: j.Name, Os: j.Os, State: j.State, ExitCode: j.ExitCode, Error: j.Error, + }) + } + return connect.NewResponse(resp), nil +} + +func (h *triggerHandler) GetJobLog(ctx context.Context, req *connect.Request[opificav1.GetJobLogRequest]) (*connect.Response[opificav1.GetJobLogResponse], error) { + repo, err := h.auth(ctx, req.Header().Get("Authorization")) + if err != nil { + return nil, err + } + id, err := uuid.Parse(req.Msg.GetJobId()) + if err != nil { + return nil, errNoSuchRun + } + job, err := h.repos.Store.GetJob(ctx, id) + if err != nil { + return nil, errNoSuchRun + } + run, err := h.repos.Store.GetRun(ctx, job.RunID) + if store.IsNotFound(err) || err == nil && (run.RepositoryID == nil || *run.RepositoryID != repo.ID) { + return nil, errNoSuchRun + } + if err != nil { + return nil, err + } + data, next, err := h.logs.ReadFrom(id, req.Msg.GetOffset(), maxLogChunk) + if err != nil { + return nil, err + } + finished := job.State != "pending" && job.State != "queued" && job.State != "assigned" && job.State != "running" + return connect.NewResponse(&opificav1.GetJobLogResponse{ + Data: data, NextOffset: next, Complete: finished && len(data) < maxLogChunk, + }), nil +} diff --git a/internal/controller/logs/logs.go b/internal/controller/logs/logs.go index 80345bb..6f2ff04 100644 --- a/internal/controller/logs/logs.go +++ b/internal/controller/logs/logs.go @@ -99,3 +99,22 @@ func (s *Store) Open(job uuid.UUID) (io.ReadCloser, error) { } return f, err } + +// ReadFrom returns up to max bytes of a job's log starting at offset, and +// the offset to continue from. +func (s *Store) ReadFrom(job uuid.UUID, offset int64, max int) ([]byte, int64, error) { + f, err := os.Open(s.path(job)) + if errors.Is(err, os.ErrNotExist) { + return nil, 0, nil + } + if err != nil { + return nil, 0, err + } + defer f.Close() + buf := make([]byte, max) + n, err := f.ReadAt(buf, offset) + if err != nil && !errors.Is(err, io.EOF) { + return nil, offset, err + } + return buf[:n], offset + int64(n), nil +} diff --git a/internal/controller/repos/repos.go b/internal/controller/repos/repos.go new file mode 100644 index 0000000..14452e1 --- /dev/null +++ b/internal/controller/repos/repos.go @@ -0,0 +1,383 @@ +// Package repos is the controller's registry of repositories: where their +// code lives, which pool each kind of event runs in, and the tokens that may +// trigger runs. +// +// Runs are triggered the same way wherever a repository is hosted: by the +// git hook Opifica installs in hosted repositories, or by `opifica trigger` in +// a GitHub Actions workflow. Either way the controller reads the pipeline +// file at the commit itself, and decides the pool from the repository's +// routing rules, never from the caller or the pipeline. +package repos + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "os" + gopath "path" + "path/filepath" + "regexp" + "strings" + + "github.com/google/uuid" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/internal/controller/scheduler" + "github.com/opifica/opifica/internal/controller/secrets" + "github.com/opifica/opifica/internal/controller/store" + "github.com/opifica/opifica/internal/controller/store/db" + "github.com/opifica/opifica/internal/gitmirror" + "github.com/opifica/opifica/internal/pipeline" +) + +// Route sends matching events to a pool. +type Route struct { + // Event is "push", "tag", "pull_request" or "*". + Event string `json:"event"` + // Ref is a glob on the branch name (push), tag name (tag) or target + // branch (pull_request). Empty matches any. + Ref string `json:"ref,omitempty"` + Pool string `json:"pool"` +} + +var ( + nameRe = regexp.MustCompile(`^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$`) + poolRe = regexp.MustCompile(`^[a-z][a-z0-9-]{0,62}$`) + shaRe = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`) + // "push:main=trusted", "tag:v*=release", "pull_request=untrusted", "*=untrusted" + routeRe = regexp.MustCompile(`^(push|tag|pull_request|\*)(?::([^=]+))?=([a-z][a-z0-9-]*)$`) +) + +// DefaultRoutes send every event to the least trusted pool. Admins add rules +// for what deserves more trust. +var DefaultRoutes = []Route{{Event: "*", Pool: "untrusted"}} + +// ParseRoute parses "event[:ref]=pool", e.g. "push:main=trusted". +func ParseRoute(s string) (Route, error) { + m := routeRe.FindStringSubmatch(s) + if m == nil { + return Route{}, fmt.Errorf("invalid route %q: want event[:ref]=pool, e.g. push:main=trusted, tag:v*=release, pull_request=untrusted or *=untrusted", s) + } + r := Route{Event: m[1], Ref: m[2], Pool: m[3]} + return r, r.check() +} + +func (r Route) check() error { + switch r.Event { + case "push", "tag", "pull_request", "*": + default: + return fmt.Errorf("route event must be push, tag, pull_request or *, not %q", r.Event) + } + if !poolRe.MatchString(r.Pool) { + return fmt.Errorf("invalid pool %q", r.Pool) + } + if _, err := gopath.Match(r.Ref, ""); err != nil { + return fmt.Errorf("invalid ref pattern %q: %v", r.Ref, err) + } + return nil +} + +func (r Route) String() string { + s := r.Event + if r.Ref != "" { + s += ":" + r.Ref + } + return s + "=" + r.Pool +} + +// Pool returns the pool for an event: the first matching route's. +func Pool(routes []Route, ev pipeline.Event) (string, bool) { + name := ev.Branch + switch ev.Type { + case pipeline.EventTag: + name = ev.Tag + case pipeline.EventPullRequest: + name = ev.BaseBranch + } + for _, r := range routes { + if r.Event != "*" && r.Event != string(ev.Type) { + continue + } + if r.Ref != "" { + if ok, _ := matchRef(r.Ref, name); !ok { + continue + } + } + return r.Pool, true + } + return "", false +} + +func matchRef(pattern, name string) (bool, error) { + if prefix, ok := strings.CutSuffix(pattern, "/**"); ok { + return name == prefix || strings.HasPrefix(name, prefix+"/"), nil + } + return gopath.Match(pattern, name) +} + +// Service manages repositories and turns triggers into runs. +type Service struct { + Store *store.Store + Key *secrets.Key + Scheduler *scheduler.Scheduler + // Mirrors is where the controller fetches repositories to read their + // pipeline files. + Mirrors *gitmirror.Mirrors +} + +// Repository is a registered repository. +type Repository struct { + ID uuid.UUID + Name string + CloneURL string + Routes []Route + HasCredential bool +} + +func fromDB(r db.Repository) (Repository, error) { + var routes []Route + if err := json.Unmarshal(r.Routes, &routes); err != nil { + return Repository{}, fmt.Errorf("repository %s: stored routes: %w", r.Name, err) + } + return Repository{ID: r.ID, Name: r.Name, CloneURL: r.CloneUrl, Routes: routes, HasCredential: r.CredentialCiphertext != nil}, nil +} + +func credentialPurpose(name string) string { return "repository-credential:" + name } + +// Create registers a repository. A nil credential means the clone URL needs +// none. Routes default to DefaultRoutes. +func (s *Service) Create(ctx context.Context, name, cloneURL string, routes []Route, credential []byte) (Repository, error) { + if !nameRe.MatchString(name) { + return Repository{}, fmt.Errorf("invalid repository name %q: want owner/name", name) + } + if cloneURL == "" { + return Repository{}, errors.New("a clone URL is required") + } + if len(routes) == 0 { + routes = DefaultRoutes + } + for _, r := range routes { + if err := r.check(); err != nil { + return Repository{}, err + } + } + routesJSON, err := json.Marshal(routes) + if err != nil { + return Repository{}, err + } + var ct, nonce []byte + if credential != nil { + if ct, nonce, err = s.Key.Seal(credentialPurpose(name), credential); err != nil { + return Repository{}, err + } + } + r, err := s.Store.CreateRepository(ctx, db.CreateRepositoryParams{ + ID: uuid.New(), Name: name, CloneUrl: cloneURL, Routes: routesJSON, CredentialCiphertext: ct, CredentialNonce: nonce, + }) + if err != nil { + if strings.Contains(err.Error(), "duplicate key") { + return Repository{}, fmt.Errorf("repository %s already exists", name) + } + return Repository{}, err + } + return fromDB(r) +} + +// Get returns a repository by name. +func (s *Service) Get(ctx context.Context, name string) (Repository, error) { + r, err := s.Store.GetRepositoryByName(ctx, name) + if store.IsNotFound(err) { + return Repository{}, fmt.Errorf("no repository named %s", name) + } + if err != nil { + return Repository{}, err + } + return fromDB(r) +} + +// List returns every repository. +func (s *Service) List(ctx context.Context) ([]Repository, error) { + rows, err := s.Store.ListRepositories(ctx) + if err != nil { + return nil, err + } + out := make([]Repository, 0, len(rows)) + for _, r := range rows { + repo, err := fromDB(r) + if err != nil { + return nil, err + } + out = append(out, repo) + } + return out, nil +} + +const tokenPrefix = "opr_" + +// CreateToken creates a trigger token for a repository. The token is shown +// once; only its hash is stored. +func (s *Service) CreateToken(ctx context.Context, repo, description string) (string, error) { + r, err := s.Get(ctx, repo) + if err != nil { + return "", err + } + raw := make([]byte, 32) + if _, err := rand.Read(raw); err != nil { + return "", err + } + token := tokenPrefix + base64.RawURLEncoding.EncodeToString(raw) + if _, err := s.Store.CreateTriggerToken(ctx, db.CreateTriggerTokenParams{ + ID: uuid.New(), RepositoryID: r.ID, TokenHash: hash(token), Description: description, + }); err != nil { + return "", err + } + return token, nil +} + +func hash(token string) []byte { + h := sha256.Sum256([]byte(token)) + return h[:] +} + +// ErrBadToken is returned for unknown or revoked trigger tokens. +var ErrBadToken = errors.New("invalid trigger token") + +// Authenticate returns the repository a trigger token belongs to. +func (s *Service) Authenticate(ctx context.Context, token string) (Repository, error) { + if !strings.HasPrefix(token, tokenPrefix) { + return Repository{}, ErrBadToken + } + t, err := s.Store.UseTriggerToken(ctx, hash(token)) + if store.IsNotFound(err) { + return Repository{}, ErrBadToken + } + if err != nil { + return Repository{}, err + } + r, err := s.Store.GetRepository(ctx, t.RepositoryID) + if err != nil { + return Repository{}, err + } + return fromDB(r) +} + +// credential returns a repository's decrypted read credential, if any. +func (s *Service) credential(ctx context.Context, repoID uuid.UUID) (*opificav1.Credential, error) { + r, err := s.Store.GetRepository(ctx, repoID) + if err != nil { + return nil, err + } + if r.CredentialCiphertext == nil { + return nil, nil + } + tok, err := s.Key.Open(credentialPurpose(r.Name), r.CredentialCiphertext, r.CredentialNonce) + if err != nil { + return nil, err + } + return &opificav1.Credential{Kind: &opificav1.Credential_Token{Token: string(tok)}}, nil +} + +// RunCredential returns the read credential for a run's repository; it +// plugs into scheduler.Config.Credentials. +func (s *Service) RunCredential(ctx context.Context, runID uuid.UUID) (*opificav1.Credential, error) { + run, err := s.Store.GetRun(ctx, runID) + if err != nil { + return nil, err + } + if run.RepositoryID == nil { + return nil, nil + } + return s.credential(ctx, *run.RepositoryID) +} + +// Outcome is what happened with a trigger. +type Outcome struct { + RunID uuid.UUID // zero if no run was started + Pool string + Jobs int + Message string +} + +// Trigger starts a run for a ref update in repo, if the routing rules and +// the pipeline say there is something to run. Problems with the request are +// errors; "nothing to do" is an Outcome with a message. +func (s *Service) Trigger(ctx context.Context, repo Repository, req *opificav1.TriggerRequest) (Outcome, error) { + ev, err := event(req) + if err != nil { + return Outcome{}, err + } + pool, ok := Pool(repo.Routes, ev) + if !ok { + return Outcome{Message: "no routing rule of " + repo.Name + " matches this event; nothing to run"}, nil + } + + cred, err := s.credential(ctx, repo.ID) + if err != nil { + return Outcome{}, err + } + mirror, err := s.Mirrors.Ensure(ctx, repo.CloneURL, cred, ev.SHA) + if err != nil { + return Outcome{}, fmt.Errorf("fetching %s: %w", repo.Name, err) + } + src, err := gitmirror.ReadFile(ctx, mirror, ev.SHA, filepath.ToSlash(pipeline.DefaultPath)) + if errors.Is(err, os.ErrNotExist) { + return Outcome{Message: "no " + pipeline.DefaultPath + " at this commit; nothing to run"}, nil + } + if err != nil { + return Outcome{}, err + } + + run, err := s.Scheduler.CreateRun(ctx, scheduler.RunRequest{ + Repo: repo.Name, CloneURL: repo.CloneURL, Event: ev, Pool: pool, Pipeline: src, RepositoryID: &repo.ID, + }) + switch { + case errors.Is(err, scheduler.ErrNotTriggered): + return Outcome{Pool: pool, Message: "the pipeline's triggers don't match this event; nothing to run"}, nil + case errors.Is(err, scheduler.ErrNothingToRun): + return Outcome{Pool: pool, Message: "every job's condition is false for this event; nothing to run"}, nil + case err != nil: + return Outcome{}, err + } + jobs, err := s.Store.ListRunJobs(ctx, run.ID) + if err != nil { + return Outcome{}, err + } + return Outcome{RunID: run.ID, Pool: pool, Jobs: len(jobs)}, nil +} + +// event turns a trigger request into a pipeline event. +func event(req *opificav1.TriggerRequest) (pipeline.Event, error) { + sha := req.GetCommitSha() + if !shaRe.MatchString(sha) { + return pipeline.Event{}, fmt.Errorf("commit_sha must be a full hex commit id, got %q", sha) + } + ref := req.GetRef() + ev := pipeline.Event{SHA: sha, Ref: ref} + switch req.GetEvent() { + case "push": + branch, ok := strings.CutPrefix(ref, "refs/heads/") + if !ok || branch == "" { + return pipeline.Event{}, fmt.Errorf("push ref must be refs/heads/, got %q", ref) + } + ev.Type, ev.Branch = pipeline.EventPush, branch + case "tag": + tag, ok := strings.CutPrefix(ref, "refs/tags/") + if !ok || tag == "" { + return pipeline.Event{}, fmt.Errorf("tag ref must be refs/tags/, got %q", ref) + } + ev.Type, ev.Tag = pipeline.EventTag, tag + case "pull_request": + if req.GetBaseBranch() == "" { + return pipeline.Event{}, errors.New("pull requests need base_branch") + } + ev.Type, ev.BaseBranch = pipeline.EventPullRequest, req.GetBaseBranch() + ev.Branch = strings.TrimPrefix(ref, "refs/heads/") + default: + return pipeline.Event{}, fmt.Errorf("event must be push, tag or pull_request, got %q", req.GetEvent()) + } + return ev, nil +} diff --git a/internal/controller/repos/repos_test.go b/internal/controller/repos/repos_test.go new file mode 100644 index 0000000..65bce54 --- /dev/null +++ b/internal/controller/repos/repos_test.go @@ -0,0 +1,92 @@ +package repos + +import ( + "strings" + "testing" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/internal/pipeline" +) + +func TestParseRoute(t *testing.T) { + for in, want := range map[string]Route{ + "push:main=trusted": {Event: "push", Ref: "main", Pool: "trusted"}, + "tag:v*=release": {Event: "tag", Ref: "v*", Pool: "release"}, + "pull_request=untrusted": {Event: "pull_request", Pool: "untrusted"}, + "*=untrusted": {Event: "*", Pool: "untrusted"}, + "push:release/**=trusted": {Event: "push", Ref: "release/**", Pool: "trusted"}, + } { + got, err := ParseRoute(in) + if err != nil || got != want { + t.Errorf("ParseRoute(%q) = %+v, %v; want %+v", in, got, err, want) + } + if got.String() != in { + t.Errorf("String() = %q, want %q", got.String(), in) + } + } + for _, bad := range []string{"", "push", "merge=trusted", "push:main=Trusted", "push:[=trusted", "push:main="} { + if _, err := ParseRoute(bad); err == nil { + t.Errorf("ParseRoute(%q) accepted", bad) + } + } +} + +func TestPool(t *testing.T) { + routes := []Route{ + {Event: "push", Ref: "main", Pool: "trusted"}, + {Event: "push", Ref: "release/**", Pool: "trusted"}, + {Event: "tag", Ref: "v*", Pool: "release"}, + {Event: "pull_request", Ref: "main", Pool: "untrusted"}, + } + tests := []struct { + ev pipeline.Event + want string + ok bool + }{ + {pipeline.Event{Type: pipeline.EventPush, Branch: "main"}, "trusted", true}, + {pipeline.Event{Type: pipeline.EventPush, Branch: "release/1.0"}, "trusted", true}, + {pipeline.Event{Type: pipeline.EventPush, Branch: "feature"}, "", false}, + {pipeline.Event{Type: pipeline.EventTag, Tag: "v2.0"}, "release", true}, + {pipeline.Event{Type: pipeline.EventTag, Tag: "nightly"}, "", false}, + // For pull requests the target branch counts, never the source: + // naming your branch "main" gets you nothing. + {pipeline.Event{Type: pipeline.EventPullRequest, Branch: "main", BaseBranch: "develop"}, "", false}, + {pipeline.Event{Type: pipeline.EventPullRequest, Branch: "feature", BaseBranch: "main"}, "untrusted", true}, + } + for _, tt := range tests { + got, ok := Pool(routes, tt.ev) + if got != tt.want || ok != tt.ok { + t.Errorf("Pool(%+v) = %q, %v; want %q, %v", tt.ev, got, ok, tt.want, tt.ok) + } + } + if got, _ := Pool(DefaultRoutes, pipeline.Event{Type: pipeline.EventTag, Tag: "v1"}); got != "untrusted" { + t.Errorf("default routes send a tag to %q", got) + } +} + +func TestEvent(t *testing.T) { + sha := strings.Repeat("a", 40) + ev, err := event(&opificav1.TriggerRequest{Event: "push", Ref: "refs/heads/main", CommitSha: sha}) + if err != nil || ev.Type != pipeline.EventPush || ev.Branch != "main" { + t.Errorf("push: %+v, %v", ev, err) + } + ev, err = event(&opificav1.TriggerRequest{Event: "tag", Ref: "refs/tags/v1", CommitSha: sha}) + if err != nil || ev.Tag != "v1" { + t.Errorf("tag: %+v, %v", ev, err) + } + ev, err = event(&opificav1.TriggerRequest{Event: "pull_request", Ref: "refs/heads/fix", CommitSha: sha, BaseBranch: "main"}) + if err != nil || ev.Branch != "fix" || ev.BaseBranch != "main" { + t.Errorf("pull request: %+v, %v", ev, err) + } + for name, req := range map[string]*opificav1.TriggerRequest{ + "short sha": {Event: "push", Ref: "refs/heads/main", CommitSha: "abc"}, + "tag as push": {Event: "push", Ref: "refs/tags/v1", CommitSha: sha}, + "branch as tag": {Event: "tag", Ref: "refs/heads/main", CommitSha: sha}, + "pr without base": {Event: "pull_request", Ref: "refs/heads/x", CommitSha: sha}, + "unknown event": {Event: "merge", Ref: "refs/heads/main", CommitSha: sha}, + } { + if _, err := event(req); err == nil { + t.Errorf("%s: accepted", name) + } + } +} diff --git a/internal/controller/scheduler/scheduler.go b/internal/controller/scheduler/scheduler.go index 2b7b523..0636621 100644 --- a/internal/controller/scheduler/scheduler.go +++ b/internal/controller/scheduler/scheduler.go @@ -53,6 +53,10 @@ type Config struct { Logs *logs.Store // Secrets answers secret requests from jobs. Nil denies them all. Secrets SecretBroker + // Credentials returns the read credential for a run's repository, added + // to each job when it is handed to a worker (never stored in the job). + // Nil means repositories need no credential. + Credentials func(ctx context.Context, run uuid.UUID) (*opificav1.Credential, error) } // SecretBroker decides secret requests; see secrets.Manager. @@ -127,6 +131,8 @@ type RunRequest struct { Pool string // Pipeline is the pipeline file as it is at Event.SHA. Pipeline []byte + // RepositoryID links the run to a registered repository, if any. + RepositoryID *uuid.UUID } // ErrNotTriggered means the event doesn't match the pipeline's triggers. @@ -163,7 +169,7 @@ func (s *Scheduler) CreateRun(ctx context.Context, req RunRequest) (db.Run, erro run, err = q.CreateRun(ctx, db.CreateRunParams{ ID: runID, Repo: req.Repo, CloneUrl: req.CloneURL, EventType: string(ev.Type), Ref: ev.Ref, CommitSha: ev.SHA, Branch: ev.Branch, BaseBranch: ev.BaseBranch, Tag: ev.Tag, - Pool: req.Pool, PipelineSource: string(req.Pipeline), + Pool: req.Pool, PipelineSource: string(req.Pipeline), RepositoryID: req.RepositoryID, }) if err != nil { return err @@ -359,6 +365,17 @@ func (s *Scheduler) assignOne(ctx context.Context, w workers.SessionInfo) bool { return true } spec.JobToken = token + if s.cfg.Credentials != nil { + cred, err := s.cfg.Credentials(ctx, job.RunID) + if err != nil { + s.log.Error("loading the repository credential; putting the job back", "job", job.ID, "err", err) + s.store.UnclaimJob(ctx, db.UnclaimJobParams{ID: job.ID, WorkerID: &w.WorkerID}) + return false + } + if cred != nil { + spec.Source.Credential = cred + } + } sendCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() diff --git a/internal/controller/secrets/secrets.go b/internal/controller/secrets/secrets.go index 02b6f48..1b8e202 100644 --- a/internal/controller/secrets/secrets.go +++ b/internal/controller/secrets/secrets.go @@ -107,6 +107,17 @@ func (k *Key) seal(name string, value []byte) (ciphertext, nonce []byte, err err return k.aead.Seal(nil, nonce, value, []byte(name)), nonce, nil } +// Seal encrypts value for purpose, e.g. "repository-credential:acme/app". +// Open only succeeds with the same purpose. +func (k *Key) Seal(purpose string, value []byte) (ciphertext, nonce []byte, err error) { + return k.seal(purpose, value) +} + +// Open decrypts a value sealed for purpose. +func (k *Key) Open(purpose string, ciphertext, nonce []byte) ([]byte, error) { + return k.open(purpose, ciphertext, nonce) +} + func (k *Key) open(name string, ciphertext, nonce []byte) ([]byte, error) { v, err := k.aead.Open(nil, nonce, ciphertext, []byte(name)) if err != nil { diff --git a/internal/controller/store/db/models.go b/internal/controller/store/db/models.go index 99a7002..5fce9f1 100644 --- a/internal/controller/store/db/models.go +++ b/internal/controller/store/db/models.go @@ -49,6 +49,17 @@ type JoinToken struct { UsedBy *uuid.UUID } +type Repository struct { + ID uuid.UUID + Name string + CloneUrl string + Routes json.RawMessage + CredentialCiphertext []byte + CredentialNonce []byte + CreatedAt time.Time + UpdatedAt time.Time +} + type Run struct { ID uuid.UUID Repo string @@ -65,6 +76,7 @@ type Run struct { CancelRequestedAt *time.Time CreatedAt time.Time FinishedAt *time.Time + RepositoryID *uuid.UUID } type Secret struct { @@ -92,6 +104,16 @@ type SecretAccess struct { At time.Time } +type TriggerToken struct { + ID uuid.UUID + RepositoryID uuid.UUID + TokenHash []byte + Description string + CreatedAt time.Time + LastUsedAt *time.Time + RevokedAt *time.Time +} + type Worker struct { ID uuid.UUID Hostname string diff --git a/internal/controller/store/db/repositories.sql.go b/internal/controller/store/db/repositories.sql.go new file mode 100644 index 0000000..fba4f7f --- /dev/null +++ b/internal/controller/store/db/repositories.sql.go @@ -0,0 +1,286 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: repositories.sql + +package db + +import ( + "context" + "encoding/json" + + "github.com/google/uuid" +) + +const createRepository = `-- name: CreateRepository :one +INSERT INTO repositories (id, name, clone_url, routes, credential_ciphertext, credential_nonce) +VALUES ($1, $2, $3, $4, $5, $6) +RETURNING id, name, clone_url, routes, credential_ciphertext, credential_nonce, created_at, updated_at +` + +type CreateRepositoryParams struct { + ID uuid.UUID + Name string + CloneUrl string + Routes json.RawMessage + CredentialCiphertext []byte + CredentialNonce []byte +} + +func (q *Queries) CreateRepository(ctx context.Context, arg CreateRepositoryParams) (Repository, error) { + row := q.db.QueryRow(ctx, createRepository, + arg.ID, + arg.Name, + arg.CloneUrl, + arg.Routes, + arg.CredentialCiphertext, + arg.CredentialNonce, + ) + var i Repository + err := row.Scan( + &i.ID, + &i.Name, + &i.CloneUrl, + &i.Routes, + &i.CredentialCiphertext, + &i.CredentialNonce, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const createTriggerToken = `-- name: CreateTriggerToken :one +INSERT INTO trigger_tokens (id, repository_id, token_hash, description) +VALUES ($1, $2, $3, $4) +RETURNING id, repository_id, token_hash, description, created_at, last_used_at, revoked_at +` + +type CreateTriggerTokenParams struct { + ID uuid.UUID + RepositoryID uuid.UUID + TokenHash []byte + Description string +} + +func (q *Queries) CreateTriggerToken(ctx context.Context, arg CreateTriggerTokenParams) (TriggerToken, error) { + row := q.db.QueryRow(ctx, createTriggerToken, + arg.ID, + arg.RepositoryID, + arg.TokenHash, + arg.Description, + ) + var i TriggerToken + err := row.Scan( + &i.ID, + &i.RepositoryID, + &i.TokenHash, + &i.Description, + &i.CreatedAt, + &i.LastUsedAt, + &i.RevokedAt, + ) + return i, err +} + +const deleteRepository = `-- name: DeleteRepository :execrows +DELETE FROM repositories WHERE name = $1 +` + +func (q *Queries) DeleteRepository(ctx context.Context, name string) (int64, error) { + result, err := q.db.Exec(ctx, deleteRepository, name) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const getRepository = `-- name: GetRepository :one +SELECT id, name, clone_url, routes, credential_ciphertext, credential_nonce, created_at, updated_at FROM repositories WHERE id = $1 +` + +func (q *Queries) GetRepository(ctx context.Context, id uuid.UUID) (Repository, error) { + row := q.db.QueryRow(ctx, getRepository, id) + var i Repository + err := row.Scan( + &i.ID, + &i.Name, + &i.CloneUrl, + &i.Routes, + &i.CredentialCiphertext, + &i.CredentialNonce, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const getRepositoryByName = `-- name: GetRepositoryByName :one +SELECT id, name, clone_url, routes, credential_ciphertext, credential_nonce, created_at, updated_at FROM repositories WHERE name = $1 +` + +func (q *Queries) GetRepositoryByName(ctx context.Context, name string) (Repository, error) { + row := q.db.QueryRow(ctx, getRepositoryByName, name) + var i Repository + err := row.Scan( + &i.ID, + &i.Name, + &i.CloneUrl, + &i.Routes, + &i.CredentialCiphertext, + &i.CredentialNonce, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const listRepositories = `-- name: ListRepositories :many +SELECT id, name, clone_url, routes, credential_ciphertext, credential_nonce, created_at, updated_at FROM repositories ORDER BY name +` + +func (q *Queries) ListRepositories(ctx context.Context) ([]Repository, error) { + rows, err := q.db.Query(ctx, listRepositories) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Repository + for rows.Next() { + var i Repository + if err := rows.Scan( + &i.ID, + &i.Name, + &i.CloneUrl, + &i.Routes, + &i.CredentialCiphertext, + &i.CredentialNonce, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listTriggerTokens = `-- name: ListTriggerTokens :many +SELECT id, repository_id, token_hash, description, created_at, last_used_at, revoked_at FROM trigger_tokens WHERE repository_id = $1 ORDER BY created_at +` + +func (q *Queries) ListTriggerTokens(ctx context.Context, repositoryID uuid.UUID) ([]TriggerToken, error) { + rows, err := q.db.Query(ctx, listTriggerTokens, repositoryID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []TriggerToken + for rows.Next() { + var i TriggerToken + if err := rows.Scan( + &i.ID, + &i.RepositoryID, + &i.TokenHash, + &i.Description, + &i.CreatedAt, + &i.LastUsedAt, + &i.RevokedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const revokeTriggerToken = `-- name: RevokeTriggerToken :execrows +UPDATE trigger_tokens SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL +` + +func (q *Queries) RevokeTriggerToken(ctx context.Context, id uuid.UUID) (int64, error) { + result, err := q.db.Exec(ctx, revokeTriggerToken, id) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const setRunRepository = `-- name: SetRunRepository :exec +UPDATE runs SET repository_id = $2 WHERE id = $1 +` + +type SetRunRepositoryParams struct { + ID uuid.UUID + RepositoryID *uuid.UUID +} + +func (q *Queries) SetRunRepository(ctx context.Context, arg SetRunRepositoryParams) error { + _, err := q.db.Exec(ctx, setRunRepository, arg.ID, arg.RepositoryID) + return err +} + +const updateRepository = `-- name: UpdateRepository :one +UPDATE repositories +SET clone_url = $2, routes = $3, credential_ciphertext = $4, credential_nonce = $5, updated_at = now() +WHERE name = $1 +RETURNING id, name, clone_url, routes, credential_ciphertext, credential_nonce, created_at, updated_at +` + +type UpdateRepositoryParams struct { + Name string + CloneUrl string + Routes json.RawMessage + CredentialCiphertext []byte + CredentialNonce []byte +} + +func (q *Queries) UpdateRepository(ctx context.Context, arg UpdateRepositoryParams) (Repository, error) { + row := q.db.QueryRow(ctx, updateRepository, + arg.Name, + arg.CloneUrl, + arg.Routes, + arg.CredentialCiphertext, + arg.CredentialNonce, + ) + var i Repository + err := row.Scan( + &i.ID, + &i.Name, + &i.CloneUrl, + &i.Routes, + &i.CredentialCiphertext, + &i.CredentialNonce, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const useTriggerToken = `-- name: UseTriggerToken :one +UPDATE trigger_tokens SET last_used_at = now() +WHERE token_hash = $1 AND revoked_at IS NULL +RETURNING id, repository_id, token_hash, description, created_at, last_used_at, revoked_at +` + +// Finds a live token and records its use. +func (q *Queries) UseTriggerToken(ctx context.Context, tokenHash []byte) (TriggerToken, error) { + row := q.db.QueryRow(ctx, useTriggerToken, tokenHash) + var i TriggerToken + err := row.Scan( + &i.ID, + &i.RepositoryID, + &i.TokenHash, + &i.Description, + &i.CreatedAt, + &i.LastUsedAt, + &i.RevokedAt, + ) + return i, err +} diff --git a/internal/controller/store/db/runs.sql.go b/internal/controller/store/db/runs.sql.go index 754cd4f..536cfce 100644 --- a/internal/controller/store/db/runs.sql.go +++ b/internal/controller/store/db/runs.sql.go @@ -179,9 +179,9 @@ func (q *Queries) CreateJob(ctx context.Context, arg CreateJobParams) error { } const createRun = `-- name: CreateRun :one -INSERT INTO runs (id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source) -VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) -RETURNING id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at +INSERT INTO runs (id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, repository_id) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) +RETURNING id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at, repository_id ` type CreateRunParams struct { @@ -196,6 +196,7 @@ type CreateRunParams struct { Tag string Pool string PipelineSource string + RepositoryID *uuid.UUID } func (q *Queries) CreateRun(ctx context.Context, arg CreateRunParams) (Run, error) { @@ -211,6 +212,7 @@ func (q *Queries) CreateRun(ctx context.Context, arg CreateRunParams) (Run, erro arg.Tag, arg.Pool, arg.PipelineSource, + arg.RepositoryID, ) var i Run err := row.Scan( @@ -229,6 +231,7 @@ func (q *Queries) CreateRun(ctx context.Context, arg CreateRunParams) (Run, erro &i.CancelRequestedAt, &i.CreatedAt, &i.FinishedAt, + &i.RepositoryID, ) return i, err } @@ -331,7 +334,7 @@ WHERE r.id = $1 AND r.state = 'running' AND NOT EXISTS ( SELECT 1 FROM jobs j WHERE j.run_id = r.id AND j.state IN ('pending', 'queued', 'assigned', 'running')) -RETURNING id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at +RETURNING id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at, repository_id ` // Closes the run once no job can change state any more. @@ -354,6 +357,7 @@ func (q *Queries) FinishRunIfDone(ctx context.Context, id uuid.UUID) (Run, error &i.CancelRequestedAt, &i.CreatedAt, &i.FinishedAt, + &i.RepositoryID, ) return i, err } @@ -394,7 +398,7 @@ func (q *Queries) GetJob(ctx context.Context, id uuid.UUID) (Job, error) { } const getRun = `-- name: GetRun :one -SELECT id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at FROM runs WHERE id = $1 +SELECT id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at, repository_id FROM runs WHERE id = $1 ` func (q *Queries) GetRun(ctx context.Context, id uuid.UUID) (Run, error) { @@ -416,6 +420,7 @@ func (q *Queries) GetRun(ctx context.Context, id uuid.UUID) (Run, error) { &i.CancelRequestedAt, &i.CreatedAt, &i.FinishedAt, + &i.RepositoryID, ) return i, err } @@ -502,7 +507,7 @@ func (q *Queries) ListRunJobs(ctx context.Context, runID uuid.UUID) ([]Job, erro } const listRuns = `-- name: ListRuns :many -SELECT id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at FROM runs ORDER BY created_at DESC LIMIT $1 +SELECT id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at, repository_id FROM runs ORDER BY created_at DESC LIMIT $1 ` func (q *Queries) ListRuns(ctx context.Context, limit int32) ([]Run, error) { @@ -530,6 +535,7 @@ func (q *Queries) ListRuns(ctx context.Context, limit int32) ([]Run, error) { &i.CancelRequestedAt, &i.CreatedAt, &i.FinishedAt, + &i.RepositoryID, ); err != nil { return nil, err } @@ -665,7 +671,7 @@ func (q *Queries) RenewLeases(ctx context.Context, arg RenewLeasesParams) error const requestRunCancel = `-- name: RequestRunCancel :one UPDATE runs SET cancel_requested_at = now() WHERE id = $1 AND state = 'running' AND cancel_requested_at IS NULL -RETURNING id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at +RETURNING id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, state, cancel_requested_at, created_at, finished_at, repository_id ` func (q *Queries) RequestRunCancel(ctx context.Context, id uuid.UUID) (Run, error) { @@ -687,6 +693,7 @@ func (q *Queries) RequestRunCancel(ctx context.Context, id uuid.UUID) (Run, erro &i.CancelRequestedAt, &i.CreatedAt, &i.FinishedAt, + &i.RepositoryID, ) return i, err } diff --git a/internal/controller/store/migrations/0004_repositories.sql b/internal/controller/store/migrations/0004_repositories.sql new file mode 100644 index 0000000..90eb45c --- /dev/null +++ b/internal/controller/store/migrations/0004_repositories.sql @@ -0,0 +1,29 @@ +-- Repositories Opifica builds: hosted on our own git server or elsewhere +-- (e.g. GitHub, triggered from a workflow). +CREATE TABLE repositories ( + id uuid PRIMARY KEY, + name text NOT NULL UNIQUE CHECK (name ~ '^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$'), + -- Where the controller and workers fetch the code. + clone_url text NOT NULL, + -- Ordered routing rules: [{"event": "push", "ref": "main", "pool": "trusted"}, ...]. + -- The first rule that matches an event decides its pool; no match, no run. + routes jsonb NOT NULL, + -- Optional read token for cloning, encrypted with the secrets key. + credential_ciphertext bytea, + credential_nonce bytea, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); + +-- Tokens that may trigger runs for one repository. Only hashes are stored. +CREATE TABLE trigger_tokens ( + id uuid PRIMARY KEY, + repository_id uuid NOT NULL REFERENCES repositories (id) ON DELETE CASCADE, + token_hash bytea NOT NULL UNIQUE, + description text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now(), + last_used_at timestamptz, + revoked_at timestamptz +); + +ALTER TABLE runs ADD COLUMN repository_id uuid REFERENCES repositories (id) ON DELETE SET NULL; diff --git a/internal/controller/store/queries/repositories.sql b/internal/controller/store/queries/repositories.sql new file mode 100644 index 0000000..cc372d0 --- /dev/null +++ b/internal/controller/store/queries/repositories.sql @@ -0,0 +1,42 @@ +-- name: CreateRepository :one +INSERT INTO repositories (id, name, clone_url, routes, credential_ciphertext, credential_nonce) +VALUES ($1, $2, $3, $4, $5, $6) +RETURNING *; + +-- name: UpdateRepository :one +UPDATE repositories +SET clone_url = $2, routes = $3, credential_ciphertext = $4, credential_nonce = $5, updated_at = now() +WHERE name = $1 +RETURNING *; + +-- name: GetRepositoryByName :one +SELECT * FROM repositories WHERE name = $1; + +-- name: GetRepository :one +SELECT * FROM repositories WHERE id = $1; + +-- name: ListRepositories :many +SELECT * FROM repositories ORDER BY name; + +-- name: DeleteRepository :execrows +DELETE FROM repositories WHERE name = $1; + +-- name: CreateTriggerToken :one +INSERT INTO trigger_tokens (id, repository_id, token_hash, description) +VALUES ($1, $2, $3, $4) +RETURNING *; + +-- name: UseTriggerToken :one +-- Finds a live token and records its use. +UPDATE trigger_tokens SET last_used_at = now() +WHERE token_hash = $1 AND revoked_at IS NULL +RETURNING *; + +-- name: ListTriggerTokens :many +SELECT * FROM trigger_tokens WHERE repository_id = $1 ORDER BY created_at; + +-- name: RevokeTriggerToken :execrows +UPDATE trigger_tokens SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL; + +-- name: SetRunRepository :exec +UPDATE runs SET repository_id = $2 WHERE id = $1; diff --git a/internal/controller/store/queries/runs.sql b/internal/controller/store/queries/runs.sql index cd794ba..0ed3b28 100644 --- a/internal/controller/store/queries/runs.sql +++ b/internal/controller/store/queries/runs.sql @@ -1,6 +1,6 @@ -- name: CreateRun :one -INSERT INTO runs (id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source) -VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) +INSERT INTO runs (id, repo, clone_url, event_type, ref, commit_sha, branch, base_branch, tag, pool, pipeline_source, repository_id) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) RETURNING *; -- name: GetRun :one diff --git a/pkg/worker/mirror.go b/internal/gitmirror/gitmirror.go similarity index 85% rename from pkg/worker/mirror.go rename to internal/gitmirror/gitmirror.go index 8e2c2c4..3b82ab9 100644 --- a/pkg/worker/mirror.go +++ b/internal/gitmirror/gitmirror.go @@ -1,4 +1,6 @@ -package worker +// Package gitmirror keeps bare mirrors of repositories on local disk. Workers +// check jobs out from them, and the controller reads pipeline files from them. +package gitmirror import ( "bytes" @@ -158,3 +160,19 @@ func runGit(ctx context.Context, env []string, args ...string) (string, error) { } return stdout.String(), nil } + +// ReadFile returns a file's content at a commit of a mirror returned by +// Ensure. It returns os.ErrNotExist if the commit has no such file. +func ReadFile(ctx context.Context, mirror, sha, file string) ([]byte, error) { + if !commitRe.MatchString(sha) { + return nil, fmt.Errorf("invalid commit id %q", sha) + } + if _, err := runGit(ctx, nil, "-C", mirror, "cat-file", "-e", sha+":"+file); err != nil { + return nil, fmt.Errorf("%s at %s: %w", file, sha, os.ErrNotExist) + } + out, err := runGit(ctx, nil, "-C", mirror, "show", sha+":"+file) + if err != nil { + return nil, err + } + return []byte(out), nil +} diff --git a/internal/integration/connect_test.go b/internal/integration/connect_test.go index 8009ce2..c59d4bb 100644 --- a/internal/integration/connect_test.go +++ b/internal/integration/connect_test.go @@ -22,10 +22,12 @@ import ( "github.com/opifica/opifica/internal/controller/api" "github.com/opifica/opifica/internal/controller/ca" "github.com/opifica/opifica/internal/controller/logs" + "github.com/opifica/opifica/internal/controller/repos" "github.com/opifica/opifica/internal/controller/scheduler" "github.com/opifica/opifica/internal/controller/secrets" "github.com/opifica/opifica/internal/controller/store" "github.com/opifica/opifica/internal/controller/workers" + "github.com/opifica/opifica/internal/gitmirror" "github.com/opifica/opifica/internal/server" "github.com/opifica/opifica/internal/testutil/pgtest" "github.com/opifica/opifica/pkg/worker" @@ -46,6 +48,7 @@ type env struct { sched *scheduler.Scheduler logs *logs.Store secrets *secrets.Manager + repos *repos.Service url string } @@ -75,9 +78,15 @@ func newEnv(t *testing.T) *env { t.Fatal(err) } secretMgr := &secrets.Manager{Store: st, Key: key} + var repoSvc *repos.Service sched := scheduler.New(st, svc.Sessions(), quiet(), scheduler.Config{ Lease: 400 * time.Millisecond, Interval: 20 * time.Millisecond, Logs: logStore, Secrets: secretMgr, + Credentials: func(ctx context.Context, run uuid.UUID) (*opificav1.Credential, error) { + return repoSvc.RunCredential(ctx, run) + }, }) + repoSvc = &repos.Service{Store: st, Key: key, Scheduler: sched, + Mirrors: &gitmirror.Mirrors{Dir: t.TempDir(), AllowLocal: true}} svc.Messages = sched go sched.Run(ctx) tlsCfg, err := api.TLSConfig(authority, []string{"127.0.0.1"}) @@ -90,7 +99,7 @@ func newEnv(t *testing.T) *env { } done := make(chan struct{}) go func() { - server.ServeTLS(ctx, ln, api.NewHandler(svc), tlsCfg, quiet()) + server.ServeTLS(ctx, ln, api.NewHandler(api.Deps{Workers: svc, Repos: repoSvc, Logs: logStore}), tlsCfg, quiet()) close(done) }() t.Cleanup(func() { @@ -98,7 +107,7 @@ func newEnv(t *testing.T) *env { <-done st.Close() }) - return &env{t: t, svc: svc, sched: sched, logs: logStore, secrets: secretMgr, url: "https://" + ln.Addr().String()} + return &env{t: t, svc: svc, sched: sched, logs: logStore, secrets: secretMgr, repos: repoSvc, url: "https://" + ln.Addr().String()} } func info() *opificav1.WorkerInfo { diff --git a/internal/integration/e2e_test.go b/internal/integration/e2e_test.go index fe35276..2854a9b 100644 --- a/internal/integration/e2e_test.go +++ b/internal/integration/e2e_test.go @@ -38,6 +38,11 @@ type realWorker struct { } func (e *env) realWorker(capacity int) *realWorker { + e.t.Helper() + return e.realWorkerIn("trusted", capacity) +} + +func (e *env) realWorkerIn(pool string, capacity int) *realWorker { e.t.Helper() if runtime.GOOS == "windows" { e.t.Skip("the test pipelines use a POSIX shell") @@ -45,20 +50,20 @@ func (e *env) realWorker(capacity int) *realWorker { if _, err := exec.LookPath("git"); err != nil { e.t.Skip("git is not installed") } - pool := worker.PoolConfig{ + vms := worker.PoolConfig{ Capacity: capacity, Images: map[string]worker.Image{"default": {Name: "default"}}, Warm: map[string]int{"default": 1}, } dir := e.t.TempDir() - e.joinInto("trusted", worker.Info(pool, nil), dir) + e.joinInto(pool, worker.Info(vms, nil), dir) rw := &realWorker{hv: &workertest.Hypervisor{Agent: true}, done: make(chan error, 1)} ctx, cancel := context.WithCancel(context.Background()) rw.cancel = cancel go func() { rw.done <- worker.Run(ctx, worker.Config{ - ControllerURL: e.url, StateDir: dir, Hypervisor: rw.hv, Pool: pool, + ControllerURL: e.url, StateDir: dir, Hypervisor: rw.hv, Pool: vms, AllowLocalRepos: true, HeartbeatInterval: 50 * time.Millisecond, Logger: quiet(), }) }() diff --git a/internal/integration/triggers_test.go b/internal/integration/triggers_test.go new file mode 100644 index 0000000..6a33da3 --- /dev/null +++ b/internal/integration/triggers_test.go @@ -0,0 +1,326 @@ +package integration + +import ( + "context" + "fmt" + "net/http" + "os" + "os/exec" + "path/filepath" + "regexp" + "strings" + "sync" + "testing" + + "connectrpc.com/connect" + "github.com/google/uuid" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" + "github.com/opifica/opifica/internal/controller/repos" + "github.com/opifica/opifica/pkg/worker" +) + +var ( + opificaOnce sync.Once + opificaBin string + opificaErr error +) + +// opificaBinary builds the opifica CLI once for the tests that run it. +func opificaBinary(t *testing.T) string { + t.Helper() + opificaOnce.Do(func() { + dir, err := os.MkdirTemp("", "opifica-cli-") + if err != nil { + opificaErr = err + return + } + opificaBin = filepath.Join(dir, "opifica") + out, err := exec.Command("go", "build", "-o", opificaBin, "github.com/opifica/opifica/cmd/opifica").CombinedOutput() + if err != nil { + opificaErr = fmt.Errorf("%v: %s", err, out) + } + }) + if opificaErr != nil { + t.Fatalf("building opifica: %v", opificaErr) + } + return opificaBin +} + +func (e *env) fingerprint() string { return e.svc.CA.Fingerprint() } + +// triggerClient is what `opifica trigger` uses, with a given token. +func (e *env) triggerClient(t *testing.T, token string) opificav1connect.TriggerServiceClient { + t.Helper() + cfg, err := worker.PinnedCAConfig(e.fingerprint(), "127.0.0.1") + if err != nil { + t.Fatal(err) + } + auth := connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { + return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { + req.Header().Set("Authorization", "Bearer "+token) + return next(ctx, req) + } + }) + hc := &http.Client{Transport: &http.Transport{TLSClientConfig: cfg, ForceAttemptHTTP2: true}} + return opificav1connect.NewTriggerServiceClient(hc, e.url, connect.WithInterceptors(auth)) +} + +func (e *env) registerRepo(t *testing.T, name, cloneURL string, routes ...string) string { + t.Helper() + var parsed []repos.Route + for _, r := range routes { + route, err := repos.ParseRoute(r) + if err != nil { + t.Fatal(err) + } + parsed = append(parsed, route) + } + if _, err := e.repos.Create(context.Background(), name, cloneURL, parsed, nil); err != nil { + t.Fatal(err) + } + token, err := e.repos.CreateToken(context.Background(), name, "test") + if err != nil { + t.Fatal(err) + } + return token +} + +func pipelineFor(body string) string { + return strings.ReplaceAll(body, "OS", hostOS()) +} + +const simplePipeline = `on: [push, tag, pull_request] +jobs: + build: + runs-on: OS + steps: [{ run: echo built $OPIFICA_REF }] +` + +func TestTriggerAPI(t *testing.T) { + e := newEnv(t) + e.realWorker(1) + repo, sha := gitRepo(t, map[string]string{".opifica/pipeline.yaml": pipelineFor(simplePipeline)}) + token := e.registerRepo(t, "acme/app", repo, "push:main=trusted", "tag:v*=release") + client := e.triggerClient(t, token) + ctx := context.Background() + + resp, err := client.Trigger(ctx, connect.NewRequest(&opificav1.TriggerRequest{ + Repository: "acme/app", Event: "push", Ref: "refs/heads/main", CommitSha: sha, + })) + if err != nil { + t.Fatal(err) + } + if resp.Msg.GetPool() != "trusted" || resp.Msg.GetJobs() != 1 { + t.Errorf("trigger = %+v", resp.Msg) + } + runID := uuid.MustParse(resp.Msg.GetRunId()) + if run, _ := e.waitRun(runID); run.State != "succeeded" || run.RepositoryID == nil { + t.Fatalf("run %s, repository %v", run.State, run.RepositoryID) + } + + got, err := client.GetRun(ctx, connect.NewRequest(&opificav1.GetRunRequest{RunId: runID.String()})) + if err != nil || got.Msg.GetState() != "succeeded" || len(got.Msg.GetJobs()) != 1 { + t.Fatalf("GetRun = %+v, %v", got, err) + } + lg, err := client.GetJobLog(ctx, connect.NewRequest(&opificav1.GetJobLogRequest{JobId: got.Msg.GetJobs()[0].GetId()})) + if err != nil || !strings.Contains(string(lg.Msg.GetData()), "built refs/heads/main") || !lg.Msg.GetComplete() { + t.Errorf("GetJobLog = %q complete=%v, %v", lg.Msg.GetData(), lg.Msg.GetComplete(), err) + } + + // Nothing to do: no route, or a commit without a pipeline file. + resp, err = client.Trigger(ctx, connect.NewRequest(&opificav1.TriggerRequest{ + Repository: "acme/app", Event: "push", Ref: "refs/heads/feature", CommitSha: sha, + })) + if err != nil || resp.Msg.GetRunId() != "" || !strings.Contains(resp.Msg.GetMessage(), "no routing rule") { + t.Errorf("unrouted push = %+v, %v", resp.Msg, err) + } +} + +func TestTriggerAuthorization(t *testing.T) { + e := newEnv(t) + repoA, shaA := gitRepo(t, map[string]string{".opifica/pipeline.yaml": pipelineFor(simplePipeline)}) + tokenA := e.registerRepo(t, "acme/a", repoA, "*=trusted") + tokenB := e.registerRepo(t, "acme/b", repoA, "*=trusted") + ctx := context.Background() + req := func(repo string) *connect.Request[opificav1.TriggerRequest] { + return connect.NewRequest(&opificav1.TriggerRequest{Repository: repo, Event: "push", Ref: "refs/heads/main", CommitSha: shaA}) + } + + if _, err := e.triggerClient(t, "opr_made-up").Trigger(ctx, req("acme/a")); connect.CodeOf(err) != connect.CodeUnauthenticated { + t.Errorf("unknown token: %v", err) + } + if _, err := e.triggerClient(t, tokenB).Trigger(ctx, req("acme/a")); connect.CodeOf(err) != connect.CodePermissionDenied { + t.Errorf("another repository's token: %v", err) + } + resp, err := e.triggerClient(t, tokenA).Trigger(ctx, req("acme/a")) + if err != nil { + t.Fatal(err) + } + // B's token can't read A's runs or logs either. + if _, err := e.triggerClient(t, tokenB).GetRun(ctx, connect.NewRequest(&opificav1.GetRunRequest{RunId: resp.Msg.GetRunId()})); connect.CodeOf(err) != connect.CodeNotFound { + t.Errorf("reading another repository's run: %v", err) + } + jobs, _ := e.svc.Store.ListRunJobs(ctx, uuid.MustParse(resp.Msg.GetRunId())) + if _, err := e.triggerClient(t, tokenB).GetJobLog(ctx, connect.NewRequest(&opificav1.GetJobLogRequest{JobId: jobs[0].ID.String()})); connect.CodeOf(err) != connect.CodeNotFound { + t.Errorf("reading another repository's log: %v", err) + } + // A revoked token stops working. + toks, _ := e.svc.Store.ListTriggerTokens(ctx, mustRepoID(t, e, "acme/a")) + e.svc.Store.RevokeTriggerToken(ctx, toks[0].ID) + if _, err := e.triggerClient(t, tokenA).Trigger(ctx, req("acme/a")); connect.CodeOf(err) != connect.CodeUnauthenticated { + t.Errorf("revoked token: %v", err) + } +} + +func mustRepoID(t *testing.T, e *env, name string) uuid.UUID { + r, err := e.repos.Get(context.Background(), name) + if err != nil { + t.Fatal(err) + } + return r.ID +} + +// A repository's read credential reaches the worker with each job, and is +// never stored in the job. +func TestRepositoryCredentialReachesWorker(t *testing.T) { + e := newEnv(t) + w := e.fakeWorker("w", workerOpts{}) + repo, sha := gitRepo(t, map[string]string{".opifica/pipeline.yaml": "on: push\njobs:\n a:\n runs-on: linux\n steps: [{ run: x }]\n"}) + if _, err := e.repos.Create(context.Background(), "acme/private", repo, []repos.Route{{Event: "*", Pool: "trusted"}}, []byte("ghp_read_only_token")); err != nil { + t.Fatal(err) + } + token, _ := e.repos.CreateToken(context.Background(), "acme/private", "") + resp, err := e.triggerClient(t, token).Trigger(context.Background(), connect.NewRequest(&opificav1.TriggerRequest{ + Repository: "acme/private", Event: "push", Ref: "refs/heads/main", CommitSha: sha, + })) + if err != nil { + t.Fatal(err) + } + e.waitRun(uuid.MustParse(resp.Msg.GetRunId())) + jobs := w.jobs() + if len(jobs) != 1 || jobs[0].GetSource().GetCredential().GetToken() != "ghp_read_only_token" { + t.Fatalf("worker got credential %v", jobs) + } + stored, _ := e.svc.Store.ListRunJobs(context.Background(), uuid.MustParse(resp.Msg.GetRunId())) + if strings.Contains(string(stored[0].Spec), "ghp_read_only_token") { + t.Error("the credential is stored in the job") + } + var raw []byte + e.svc.Store.Pool.QueryRow(context.Background(), "SELECT credential_ciphertext FROM repositories WHERE name = 'acme/private'").Scan(&raw) + if strings.Contains(string(raw), "ghp_read_only_token") { + t.Error("the credential is stored unencrypted") + } +} + +// The production path for hosted repositories: a bare repository with the +// post-receive hook, and a developer pushing to it. +func TestGitHookStartsRuns(t *testing.T) { + e := newEnv(t) + e.realWorkerIn("trusted", 1) + e.realWorkerIn("release", 1) + bin := opificaBinary(t) + + bare := filepath.Join(t.TempDir(), "app.git") + git(t, "", "init", "-q", "--bare", "-b", "main", bare) + token := e.registerRepo(t, "acme/app", bare, "push:main=trusted", "tag:v*=release") + hook := fmt.Sprintf("#!/bin/sh\nexport OPIFICA_CONTROLLER=%q OPIFICA_CA_FINGERPRINT=%q OPIFICA_REPO=acme/app OPIFICA_TRIGGER_TOKEN=%q\nexec %q hook post-receive\n", + e.url, e.fingerprint(), token, bin) + if err := os.WriteFile(filepath.Join(bare, "hooks", "post-receive"), []byte(hook), 0o755); err != nil { + t.Fatal(err) + } + + work := t.TempDir() + git(t, work, "clone", "-q", bare, ".") + os.MkdirAll(filepath.Join(work, ".opifica"), 0o755) + os.WriteFile(filepath.Join(work, ".opifica", "pipeline.yaml"), []byte(pipelineFor(simplePipeline)), 0o644) + git(t, work, "add", ".") + git(t, work, "commit", "-q", "-m", "add pipeline") + git(t, work, "tag", "v1.0") + + out := git(t, work, "push", "origin", "main", "v1.0") + runIDs := regexp.MustCompile(`opifica: (main|v1\.0): run (\S+) started in pool (\S+)`).FindAllStringSubmatch(out, -1) + if len(runIDs) != 2 { + t.Fatalf("push output doesn't show two runs:\n%s", out) + } + pools := map[string]string{} + for _, m := range runIDs { + pools[m[1]] = m[3] + if run, _ := e.waitRun(uuid.MustParse(m[2])); run.State != "succeeded" { + t.Errorf("run for %s: %s", m[1], run.State) + } + } + if pools["main"] != "trusted" || pools["v1.0"] != "release" { + t.Errorf("pools = %v; want main→trusted, v1.0→release", pools) + } + + // A push no route matches tells the pusher, and the push still succeeds. + git(t, work, "checkout", "-q", "-b", "feature") + out = git(t, work, "push", "origin", "feature") + if !strings.Contains(out, "opifica: feature: no routing rule") { + t.Errorf("unrouted push output:\n%s", out) + } +} + +// The GitHub Actions path: `opifica trigger --wait` in a workflow step, with +// the environment GitHub provides. +func TestTriggerFromGitHubActions(t *testing.T) { + e := newEnv(t) + e.realWorker(2) + bin := opificaBinary(t) + repo, sha := gitRepo(t, map[string]string{ + ".opifica/pipeline.yaml": pipelineFor(`on: push +jobs: + build: + runs-on: OS + steps: [{ run: echo hello from opifica }] + test: + needs: build + runs-on: OS + steps: [{ run: "if [ -f FAIL ]; then echo failing; exit 4; fi" }] +`)}) + token := e.registerRepo(t, "acme/app", repo, "push:main=trusted") + + run := func(sha string) (string, string, int) { + summary := filepath.Join(t.TempDir(), "summary.md") + cmd := exec.Command(bin, "trigger", "-wait", "-timeout", "2m") + cmd.Env = append(os.Environ(), + "GITHUB_ACTIONS=true", "GITHUB_EVENT_NAME=push", "GITHUB_REF=refs/heads/main", "GITHUB_SHA="+sha, + "GITHUB_STEP_SUMMARY="+summary, + "OPIFICA_CONTROLLER="+e.url, "OPIFICA_CA_FINGERPRINT="+e.fingerprint(), + "OPIFICA_REPO=acme/app", "OPIFICA_TRIGGER_TOKEN="+token) + out, _ := cmd.CombinedOutput() + data, _ := os.ReadFile(summary) + return string(out), string(data), cmd.ProcessState.ExitCode() + } + + out, summary, code := run(sha) + if code != 0 { + t.Fatalf("exit %d:\n%s", code, out) + } + wantIn(t, out, "started in pool trusted with 2 job(s)", "[build] hello from opifica", "succeeded") + wantIn(t, summary, "| build |", "| test |", ": succeeded") + + // A failing run fails the step. + os.WriteFile(filepath.Join(repo, "FAIL"), []byte("x"), 0o644) + git(t, repo, "add", ".") + git(t, repo, "commit", "-q", "-m", "break it") + out, _, code = run(git(t, repo, "rev-parse", "HEAD")) + if code != 1 || !strings.Contains(out, "[test] failing") || !strings.Contains(out, "(exit 4)") { + t.Errorf("failing run: exit %d:\n%s", code, out) + } +} + +func git(t *testing.T, dir string, args ...string) string { + t.Helper() + cmd := exec.Command("git", args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@example.com", + "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@example.com", "GIT_CONFIG_GLOBAL="+os.DevNull) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("git %v: %v: %s", args, err, out) + } + return strings.TrimSpace(string(out)) +} diff --git a/pkg/worker/runner.go b/pkg/worker/runner.go index 0e3dc9b..19d641e 100644 --- a/pkg/worker/runner.go +++ b/pkg/worker/runner.go @@ -19,6 +19,7 @@ import ( opificav1 "github.com/opifica/opifica/gen/opifica/v1" "github.com/opifica/opifica/internal/agent" + "github.com/opifica/opifica/internal/gitmirror" ) // checkoutDir is where the agent checks out the repository, relative to its @@ -28,7 +29,7 @@ const checkoutDir = "src" // RunnerConfig configures a Runner. type RunnerConfig struct { Pool *Pool - Mirrors *Mirrors + Mirrors *gitmirror.Mirrors // Send delivers a message to the controller. It should keep trying // across reconnects; see NewOutbox. Send func(msg *opificav1.ConnectRequest, onSent func()) @@ -232,7 +233,7 @@ func (r *Runner) execute(ctx context.Context, j *runningJob) (int32, *jobError) r.syslog(j, 0, "Fetching %s at %s", src.GetCloneUrl(), short(src.GetCommitSha())) mirror, err := r.cfg.Mirrors.Ensure(ctx, src.GetCloneUrl(), src.GetCredential(), src.GetCommitSha()) switch { - case errors.Is(err, ErrCommitNotFound): + case errors.Is(err, gitmirror.ErrCommitNotFound): return 0, failed("%v", err) case err != nil: return 0, errored("fetching the repository: %v", err) diff --git a/pkg/worker/worker.go b/pkg/worker/worker.go index b94a989..cf5dc2f 100644 --- a/pkg/worker/worker.go +++ b/pkg/worker/worker.go @@ -12,6 +12,7 @@ import ( "time" opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/internal/gitmirror" "github.com/opifica/opifica/internal/version" ) @@ -28,7 +29,7 @@ type Config struct { Labels map[string]string // GuestMirrorURL: see RunnerConfig. GuestMirrorURL func(vm VM, hostPath string) (string, error) - // AllowLocalRepos: see Mirrors.AllowLocal. Development only. + // AllowLocalRepos: see gitmirror.Mirrors.AllowLocal. Development only. AllowLocalRepos bool // HeartbeatInterval: see ClientConfig. HeartbeatInterval time.Duration @@ -89,7 +90,7 @@ func Run(ctx context.Context, cfg Config) error { var outbox *Outbox runner := NewRunner(RunnerConfig{ Pool: pool, - Mirrors: &Mirrors{Dir: filepath.Join(cfg.StateDir, "mirrors"), AllowLocal: cfg.AllowLocalRepos}, + Mirrors: &gitmirror.Mirrors{Dir: filepath.Join(cfg.StateDir, "mirrors"), AllowLocal: cfg.AllowLocalRepos}, Send: func(m *opificav1.ConnectRequest, onSent func()) { outbox.Send(m, onSent) }, GuestMirrorURL: cfg.GuestMirrorURL, Logger: log, diff --git a/proto/opifica/v1/trigger.proto b/proto/opifica/v1/trigger.proto new file mode 100644 index 0000000..c8c0212 --- /dev/null +++ b/proto/opifica/v1/trigger.proto @@ -0,0 +1,79 @@ +syntax = "proto3"; + +package opifica.v1; + +option go_package = "github.com/opifica/opifica/gen/opifica/v1;opificav1"; + +// TriggerService starts runs when code changes. It is called by the git +// hook Opifica installs in hosted repositories, and by `opifica trigger` in a +// GitHub Actions workflow for repositories hosted on GitHub. +// +// Every call carries a repository's trigger token (Authorization: Bearer), +// which only works for that repository. +service TriggerService { + // Trigger reports that a ref moved. The controller reads the pipeline file + // at the commit, routes the run to a pool with the repository's rules and + // schedules it. + rpc Trigger(TriggerRequest) returns (TriggerResponse); + + // GetRun returns a run's state and jobs, e.g. for `opifica trigger --wait`. + rpc GetRun(GetRunRequest) returns (GetRunResponse); + + // GetJobLog returns a job's output from an offset, so callers can follow it. + rpc GetJobLog(GetJobLogRequest) returns (GetJobLogResponse); +} + +message TriggerRequest { + // Registered repository name, e.g. "acme/app". + string repository = 1; + // "push", "tag" or "pull_request". + string event = 2; + // Full ref: refs/heads/main, refs/tags/v1.0. For pull requests, the ref of + // the head branch. + string ref = 3; + // The exact commit to build. + string commit_sha = 4; + // Pull requests only: the branch the pull request targets. + string base_branch = 5; +} + +message TriggerResponse { + // Empty when no run was started; message says why. + string run_id = 1; + string pool = 2; + int32 jobs = 3; + string message = 4; +} + +message GetRunRequest { + string run_id = 1; +} + +message GetRunResponse { + // "running", "succeeded", "failed" or "cancelled". + string state = 1; + repeated JobSummary jobs = 2; +} + +message JobSummary { + string id = 1; + string name = 2; + string os = 3; + string state = 4; + optional int32 exit_code = 5; + string error = 6; +} + +message GetJobLogRequest { + string job_id = 1; + // Byte offset to read from. + int64 offset = 2; +} + +message GetJobLogResponse { + bytes data = 1; + // Offset to ask for next time. + int64 next_offset = 2; + // True once the job has finished and data reaches the end of its log. + bool complete = 3; +}