diff --git a/agents/manager/downloader.go b/agents/manager/downloader.go index f1892bd5..1a9a3714 100644 --- a/agents/manager/downloader.go +++ b/agents/manager/downloader.go @@ -4,6 +4,8 @@ import ( "archive/tar" "compress/gzip" "context" + "crypto/sha256" + "encoding/hex" "fmt" "io" "net" @@ -86,40 +88,28 @@ var agentDownloadClient = &http.Client{ }, } -// fetchRelease issues the asset GET under the caller's context, so cancelling -// it aborts the transfer at any point — including mid-body, which no timeout -// on the request can cover. -// -// It does not check the host: callers must pass a URL they have already put -// through ValidURL, as Download does. The check lives there rather than here -// because the tests that cover the stall and cancellation paths have to reach -// a local server. -func fetchRelease(ctx context.Context, releaseURL string) (*http.Response, error) { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, releaseURL, nil) - if err != nil { - return nil, err - } - return agentDownloadClient.Do(req) -} - -// writeArchive streams body into f and closes it. extractTarGz reopens the -// archive by path, so the write handle must not outlive the copy — every -// return here closes it, including the ones that fail. -func writeArchive(f *os.File, body io.Reader, size int64) error { +// writeArchive streams body into f, digesting it on the way, and closes it. +// extractTarGz reopens the archive by path, so the write handle must not +// outlive the copy — every return here closes it, including the ones that fail. +func writeArchive(f *os.File, body io.Reader, size int64) (digest string, err error) { defer func() { _ = f.Close() }() bar := pb.Full.Start64(size) bar.Set(pb.Bytes, true) // Display in bytes instead of default kilobytes - if _, err := io.Copy(bar.NewProxyWriter(f), body); err != nil { - return err + hash := sha256.New() + if _, err := io.Copy(io.MultiWriter(bar.NewProxyWriter(f), hash), body); err != nil { + return "", err } bar.Finish() // Closed here rather than only by the defer: a write error that surfaces // at close must reach the caller instead of appearing after the archive // has already been extracted. - return f.Close() + if err := f.Close(); err != nil { + return "", err + } + return hex.EncodeToString(hash.Sum(nil)), nil } func Download(ctx context.Context, p *resources.Agent) error { @@ -141,11 +131,11 @@ func Download(ctx context.Context, p *resources.Agent) error { w.Info(fmt.Sprintf("Downloading agent %s", p.Identifier())) w.Debug("downloading", wool.Field("agent", p.Identifier()), wool.Field("url", releaseURL).Debug()) - body, size, err := openAgentRelease(ctx, releaseURL) + asset, err := openAgentRelease(ctx, releaseURL) if err != nil { return w.Wrapf(err, "cannot download agent") } - defer body.Close() + defer asset.Body.Close() tmp, err := os.CreateTemp("", "agent-*.tar.gz") if err != nil { @@ -158,9 +148,17 @@ func Download(ctx context.Context, p *resources.Agent) error { w.Error("cannot remove temp file", wool.ErrField(err)) } }(tmp.Name()) - if err = writeArchive(tmp, body, size); err != nil { + archiveDigest, err := writeArchive(tmp, asset.Body, asset.Size) + if err != nil { return w.Wrapf(err, "cannot copy agent") } + warning, err := checkArchiveDigest(p.Identifier(), asset, archiveDigest) + if err != nil { + return w.Wrap(err) + } + if warning != "" { + w.Warn(warning, wool.Field("url", releaseURL)) + } tmpDir, err := os.MkdirTemp("", "agent-*") if err != nil { @@ -196,6 +194,20 @@ func Download(ctx context.Context, p *resources.Agent) error { return nil } +// checkArchiveDigest holds downloaded bytes to the digest the release records +// for them, before anything is extracted. Without a digest nothing can be held, +// and that is said rather than passed over: the archive is then trusted on its +// name alone, which is the state every install was in before this check. +func checkArchiveDigest(identifier string, asset *ReleaseAsset, archiveDigest string) (warning string, err error) { + switch { + case asset.Digest == "": + return "release publishes no digest for this archive; installing it unverified", nil + case archiveDigest != asset.Digest: + return "", fmt.Errorf("refusing to install %s: the downloaded archive is sha256:%s but the release's %s record says sha256:%s", identifier, archiveDigest, asset.DigestSource, asset.Digest) + } + return "", nil +} + // extractTarGz decompresses gzip then extracts the tar archive at src into // dst. Replaces mholt/archiver.Unarchive (which had two unpatched path // traversal CVEs). Only .tar.gz is supported — this is all GitHub release diff --git a/agents/manager/downloader_test.go b/agents/manager/downloader_test.go index 223b9c6e..18c6c937 100644 --- a/agents/manager/downloader_test.go +++ b/agents/manager/downloader_test.go @@ -14,6 +14,7 @@ import ( "time" "github.com/codefly-dev/core/resources" + "github.com/google/go-github/v89/github" "github.com/stretchr/testify/require" ) @@ -166,19 +167,25 @@ func TestFetchReleaseAbortsWhenContextIsCancelledBeforeHeaders(t *testing.T) { // request it never returns at all, and this must report that as a failure // rather than hang the package until the suite-wide timeout. type result struct { - resp *http.Response + resp io.ReadCloser err error } returned := make(chan result, 1) go func() { - resp, err := fetchRelease(ctx, server.URL) + endpoint := server.URL + "/" + client, err := newGitHubReleaseClient().Clone(github.WithURLs(&endpoint, &endpoint)) + if err != nil { + returned <- result{err: err} + return + } + resp, _, err := client.Repositories.DownloadReleaseAsset(ctx, "owner", "repo", 1, agentDownloadClient) returned <- result{resp, err} }() select { case got := <-returned: if got.err == nil { - got.resp.Body.Close() + got.resp.Close() t.Fatal("a server that never answers must not return a response") } if !errors.Is(got.err, context.DeadlineExceeded) { @@ -199,21 +206,24 @@ func TestFetchReleaseAbortsWhenContextIsCancelledMidTransfer(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - resp, err := fetchRelease(ctx, server.URL) + endpoint := server.URL + "/" + client, err := newGitHubReleaseClient().Clone(github.WithURLs(&endpoint, &endpoint)) + require.NoError(t, err) + resp, _, err := client.Repositories.DownloadReleaseAsset(ctx, "owner", "repo", 1, agentDownloadClient) if err != nil { t.Fatal(err) } - defer resp.Body.Close() + defer resp.Close() first := make([]byte, 1) - if _, err := io.ReadFull(resp.Body, first); err != nil { + if _, err := io.ReadFull(resp, first); err != nil { t.Fatalf("expected the transfer to start: %v", err) } cancel() failed := make(chan error, 1) go func() { - _, err := io.Copy(io.Discard, resp.Body) + _, err := io.Copy(io.Discard, resp) failed <- err }() select { @@ -315,7 +325,7 @@ func TestWriteArchiveClosesItsHandle(t *testing.T) { t.Fatal(err) } - if err := writeArchive(f, strings.NewReader(payload), int64(len(payload))); err != nil { + if _, err := writeArchive(f, strings.NewReader(payload), int64(len(payload))); err != nil { t.Fatal(err) } if err := f.Close(); !errors.Is(err, os.ErrClosed) { @@ -338,7 +348,7 @@ func TestWriteArchiveClosesItsHandleWhenTheTransferFails(t *testing.T) { t.Fatal(err) } - err = writeArchive(f, errReader{err: context.Canceled}, 1024) + _, err = writeArchive(f, errReader{err: context.Canceled}, 1024) if !errors.Is(err, context.Canceled) { t.Fatalf("got %v, want the transfer failure to reach the caller", err) } @@ -346,3 +356,21 @@ func TestWriteArchiveClosesItsHandleWhenTheTransferFails(t *testing.T) { t.Fatalf("second Close returned %v — a failed transfer leaked the descriptor", err) } } + +// TestCheckArchiveDigestHoldsBytesToTheReleaseRecord: a digest mismatch is a +// refusal naming both digests and whose record was used; no digest at all is a +// warning that says the install is unverified, never a silent pass. +func TestCheckArchiveDigestHoldsBytesToTheReleaseRecord(t *testing.T) { + warning, err := checkArchiveDigest("codefly.dev/parse:0.1.1", &ReleaseAsset{Digest: "aa", DigestSource: DigestFromChecksums}, "aa") + require.NoError(t, err) + require.Empty(t, warning) + + _, err = checkArchiveDigest("codefly.dev/parse:0.1.1", &ReleaseAsset{Digest: "aa", DigestSource: DigestFromChecksums}, "bb") + require.ErrorContains(t, err, "refusing to install codefly.dev/parse:0.1.1") + require.ErrorContains(t, err, "sha256:bb") + require.ErrorContains(t, err, "checksums record says sha256:aa") + + warning, err = checkArchiveDigest("codefly.dev/parse:0.1.1", &ReleaseAsset{}, "bb") + require.NoError(t, err) + require.Contains(t, warning, "unverified") +} diff --git a/agents/manager/manager.go b/agents/manager/manager.go index 9cf843f3..9a3f2562 100644 --- a/agents/manager/manager.go +++ b/agents/manager/manager.go @@ -20,16 +20,11 @@ import ( // API access is capped at 60 requests/hour per IP, so resolving "latest" for // several agents can flakily 403; a token raises the cap to 5000/hour. func newGitHubReleaseClient() *github.Client { - token := githubReleaseToken() - var options []github.ClientOptionsFunc - if token == "" { - client, err := github.NewClient() - if err != nil { - panic(fmt.Sprintf("configure GitHub release client: %v", err)) - } - return client + token := GitHubReleaseToken() + options := []github.ClientOptionsFunc{github.WithHTTPClient(agentDownloadClient)} + if token != "" { + options = append(options, github.WithAuthToken(token)) } - options = append(options, github.WithAuthToken(token)) client, err := github.NewClient(options...) if err != nil { panic(fmt.Sprintf("configure authenticated GitHub release client: %v", err)) @@ -37,7 +32,13 @@ func newGitHubReleaseClient() *github.Client { return client } -func githubReleaseToken() string { +// GitHubReleaseToken is the one credential rule the agent loader applies when +// it downloads a release: GITHUB_TOKEN, else GH_TOKEN, else nothing. It never +// consults a `gh` CLI login, because the hosts that install agents — CI runners +// and hosted builds — have none. A caller that checks whether a release is +// installable must check with exactly this credential: verifying with a wider +// one establishes reachability for the laptop and nothing else. +func GitHubReleaseToken() string { if token := strings.TrimSpace(os.Getenv("GITHUB_TOKEN")); token != "" { return token } diff --git a/agents/manager/release_asset.go b/agents/manager/release_asset.go index d0f835b5..07f85aad 100644 --- a/agents/manager/release_asset.go +++ b/agents/manager/release_asset.go @@ -1,7 +1,11 @@ package manager import ( + "bufio" + "bytes" "context" + "encoding/hex" + "errors" "fmt" "io" "net/http" @@ -11,69 +15,174 @@ import ( "github.com/google/go-github/v89/github" ) +// ReleaseAsset is an opened loader archive together with the digest that +// establishes its bytes. +type ReleaseAsset struct { + Body io.ReadCloser + Size int64 + // Digest is the archive's expected SHA-256 (hex, no prefix) and + // DigestSource names who computed it: DigestFromChecksums when it comes + // from the checksums file the release's producer published, which is the + // only record independent of GitHub; DigestFromGitHub when only GitHub's + // own record of the upload exists; "" when the release offers nothing to + // check against, in which case Digest is also empty. + Digest string + DigestSource string +} + +// The producers a ReleaseAsset digest can come from. +const ( + DigestFromChecksums = "checksums" + DigestFromGitHub = "github" +) + // OpenReleaseAsset opens the exact GitHub release asset named by a loader URL. // The supplied client authenticates metadata and asset API requests. Redirected // bytes use the bounded, unauthenticated download client so API credentials never // travel to the asset CDN. The caller must close the returned reader. func OpenReleaseAsset(ctx context.Context, releaseURL string, client *github.Client) (io.ReadCloser, int64, error) { + asset, err := FetchReleaseAsset(ctx, releaseURL, client) + if err != nil { + return nil, 0, err + } + return asset.Body, asset.Size, nil +} + +// FetchReleaseAsset is OpenReleaseAsset with the digest the release records for +// the asset, so a caller that installs the bytes can check them. The caller +// must close Body. +func FetchReleaseAsset(ctx context.Context, releaseURL string, client *github.Client) (*ReleaseAsset, error) { u, err := url.Parse(releaseURL) if err != nil || u.Scheme != "https" || u.Host != "github.com" || u.User != nil || u.RawQuery != "" || u.Fragment != "" { - return nil, 0, fmt.Errorf("invalid GitHub release asset URL") + return nil, fmt.Errorf("invalid GitHub release asset URL") } parts := strings.Split(strings.TrimPrefix(u.Path, "/"), "/") if len(parts) != 6 || parts[2] != "releases" || parts[3] != "download" { - return nil, 0, fmt.Errorf("invalid GitHub release asset path") + return nil, fmt.Errorf("invalid GitHub release asset path") } for _, part := range parts { if part == "" || part == "." || part == ".." || strings.ContainsAny(part, "\\\r\n") { - return nil, 0, fmt.Errorf("invalid GitHub release asset path") + return nil, fmt.Errorf("invalid GitHub release asset path") } } if client == nil { - return nil, 0, fmt.Errorf("GitHub release client is required") + return nil, fmt.Errorf("GitHub release client is required") } owner, repo, tag, name := parts[0], parts[1], parts[4], parts[5] release, _, err := client.Repositories.GetReleaseByTag(ctx, owner, repo, tag) if err != nil { - return nil, 0, fmt.Errorf("resolve agent release: %w", err) + return nil, fmt.Errorf("resolve agent release: %w", err) } if release.GetDraft() || release.GetTagName() != tag { - return nil, 0, fmt.Errorf("agent release is not published at %s", tag) + return nil, fmt.Errorf("agent release is not published at %s", tag) } - var found *github.ReleaseAsset + var found, checksums *github.ReleaseAsset for _, asset := range release.Assets { - if asset.GetName() != name { - continue - } - if found != nil { - return nil, 0, fmt.Errorf("agent release has duplicate asset %s", name) + switch { + case asset.GetName() == name: + if found != nil { + return nil, fmt.Errorf("agent release has duplicate asset %s", name) + } + found = asset + case strings.HasSuffix(asset.GetName(), "checksums.txt"): + if checksums != nil { + return nil, fmt.Errorf("agent release publishes more than one checksums file (%s, %s); which one records %s is ambiguous", checksums.GetName(), asset.GetName(), name) + } + checksums = asset } - found = asset } if found == nil || found.GetID() <= 0 || found.GetBrowserDownloadURL() != releaseURL { - return nil, 0, fmt.Errorf("agent release does not contain the requested asset URL") + return nil, fmt.Errorf("agent release does not contain the requested asset URL") + } + digest, source, err := releaseAssetDigest(ctx, client, owner, repo, found, checksums) + if err != nil { + return nil, err } body, _, err := client.Repositories.DownloadReleaseAsset(ctx, owner, repo, found.GetID(), agentDownloadClient) if err != nil { - return nil, 0, fmt.Errorf("open agent release asset: %w", err) + return nil, fmt.Errorf("open agent release asset: %w", err) } if body == nil { - return nil, 0, fmt.Errorf("agent release asset returned no content") + return nil, fmt.Errorf("agent release asset returned no content") + } + return &ReleaseAsset{Body: body, Size: int64(found.GetSize()), Digest: digest, DigestSource: source}, nil +} + +// releaseAssetDigest prefers the digest the release's producer published over +// the one GitHub recorded: comparing served bytes to GitHub's own record only +// shows GitHub agreeing with itself. +func releaseAssetDigest(ctx context.Context, client *github.Client, owner, repo string, asset, checksums *github.ReleaseAsset) (string, string, error) { + if checksums != nil { + body, _, err := client.Repositories.DownloadReleaseAsset(ctx, owner, repo, checksums.GetID(), agentDownloadClient) + if err != nil { + return "", "", fmt.Errorf("read the release's checksums: %w", err) + } + if body == nil { + return "", "", fmt.Errorf("release checksums returned no content") + } + defer func() { _ = body.Close() }() + var raw bytes.Buffer + if _, err := io.Copy(&raw, io.LimitReader(body, (1<<20)+1)); err != nil { + return "", "", fmt.Errorf("read the release's checksums: %w", err) + } + if raw.Len() > 1<<20 { + return "", "", fmt.Errorf("release checksums exceed 1 MiB") + } + entries, err := parseChecksums(raw.Bytes()) + if err != nil { + return "", "", err + } + if digest, ok := entries[asset.GetName()]; ok { + return digest, DigestFromChecksums, nil + } + return "", "", fmt.Errorf("the release's %s records no digest for %s, so the archive cannot be established against its producer's record", checksums.GetName(), asset.GetName()) } - return body, int64(found.GetSize()), nil + if digest := asset.GetDigest(); digest != "" { + recorded, ok := strings.CutPrefix(digest, "sha256:") + if !ok || !validSHA256(recorded) { + return "", "", fmt.Errorf("release asset has an invalid SHA-256 digest") + } + return strings.ToLower(recorded), DigestFromGitHub, nil + } + return "", "", nil +} + +func validSHA256(value string) bool { + decoded, err := hex.DecodeString(value) + return err == nil && len(decoded) == 32 } -func openAgentRelease(ctx context.Context, releaseURL string) (io.ReadCloser, int64, error) { - if githubReleaseToken() != "" { - return OpenReleaseAsset(ctx, releaseURL, newGitHubReleaseClient()) +// parseChecksums reads SHA-256 records and refuses malformed or duplicate entries. +func parseChecksums(raw []byte) (map[string]string, error) { + checksums := map[string]string{} + scanner := bufio.NewScanner(bytes.NewReader(raw)) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) == 0 { + continue + } + if len(fields) != 2 || !validSHA256(fields[0]) { + return nil, fmt.Errorf("release checksums contain an invalid SHA-256 entry") + } + name := strings.TrimPrefix(fields[1], "*") + if _, exists := checksums[name]; exists { + return nil, fmt.Errorf("release checksums contain duplicate entry %q", name) + } + checksums[name] = strings.ToLower(fields[0]) } - resp, err := fetchRelease(ctx, releaseURL) - if err != nil { - return nil, 0, err + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("read release checksums: %w", err) } - if resp.StatusCode != http.StatusOK { - resp.Body.Close() - return nil, 0, fmt.Errorf("unexpected status code %d when downloading agent", resp.StatusCode) + return checksums, nil +} + +// openAgentRelease uses the same release metadata and digest discovery for +// anonymous and authenticated downloads. Only credential selection differs. +func openAgentRelease(ctx context.Context, releaseURL string) (*ReleaseAsset, error) { + asset, err := FetchReleaseAsset(ctx, releaseURL, newGitHubReleaseClient()) + var response *github.ErrorResponse + if GitHubReleaseToken() == "" && errors.As(err, &response) && response.Response.StatusCode == http.StatusNotFound { + return nil, fmt.Errorf("%w: the release may be missing, or private; set GITHUB_TOKEN or GH_TOKEN for private agent downloads (a gh CLI login is not read here)", err) } - return resp.Body, resp.ContentLength, nil + return asset, err } diff --git a/agents/manager/release_asset_test.go b/agents/manager/release_asset_test.go index ba86b47d..71ff27a9 100644 --- a/agents/manager/release_asset_test.go +++ b/agents/manager/release_asset_test.go @@ -117,3 +117,91 @@ func TestOpenReleaseAssetRefusesUntrustedURLAndCancelledContext(t *testing.T) { _, _, err = OpenReleaseAsset(ctx, testReleaseURL, client) require.ErrorIs(t, err, context.Canceled) } + +// TestFetchReleaseAssetPrefersTheProducersChecksums establishes the digest +// source order: the checksums file the release's producer published outranks +// GitHub's own record of the upload, and a release with neither yields nothing +// to hold the bytes to rather than an invented digest. +func TestFetchReleaseAssetPrefersTheProducersChecksums(t *testing.T) { + const archive = "service-render_0.1.0_linux_amd64.tar.gz" + for _, scenario := range []struct { + name, wantDigest, wantSource string + wantError string + checksums bool + githubDigest string + }{ + {name: "checksums outrank github", checksums: true, githubDigest: "sha256:" + strings.Repeat("b", 64), wantDigest: strings.Repeat("a", 64), wantSource: DigestFromChecksums}, + {name: "github record alone", githubDigest: "sha256:" + strings.Repeat("b", 64), wantDigest: strings.Repeat("b", 64), wantSource: DigestFromGitHub}, + {name: "nothing to hold to"}, + {name: "empty digest", githubDigest: "sha256:", wantError: "invalid SHA-256"}, + {name: "unsupported digest", githubDigest: "sha512:" + strings.Repeat("b", 128), wantError: "invalid SHA-256"}, + {name: "malformed digest", githubDigest: "sha256:bad", wantError: "invalid SHA-256"}, + } { + t.Run(scenario.name, func(t *testing.T) { + assets := []map[string]any{{"id": 12, "name": archive, "size": 7, "browser_download_url": testReleaseURL, "digest": scenario.githubDigest}} + if scenario.checksums { + assets = append(assets, map[string]any{"id": 13, "name": "service-render_0.1.0_checksums.txt", "size": 80, "browser_download_url": strings.Replace(testReleaseURL, archive, "service-render_0.1.0_checksums.txt", 1)}) + } + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/repos/example-org/service-render/releases/tags/v0.1.0": + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"tag_name": "v0.1.0", "assets": assets}) + case "/repos/example-org/service-render/releases/assets/13": + fmt.Fprintf(w, "%s %s\n%s other.tar.gz\n", strings.Repeat("a", 64), archive, strings.Repeat("c", 64)) + case "/repos/example-org/service-render/releases/assets/12": + fmt.Fprint(w, "archive") + default: + http.NotFound(w, r) + } + })) + t.Cleanup(api.Close) + endpoint := api.URL + "/" + client, err := github.NewClient(github.WithURLs(&endpoint, &endpoint)) + require.NoError(t, err) + asset, err := FetchReleaseAsset(t.Context(), testReleaseURL, client) + if scenario.wantError != "" { + require.ErrorContains(t, err, scenario.wantError) + return + } + require.NoError(t, err) + defer asset.Body.Close() + require.Equal(t, scenario.wantDigest, asset.Digest) + require.Equal(t, scenario.wantSource, asset.DigestSource) + }) + } +} + +// TestFetchReleaseAssetRefusesAChecksumsFileThatOmitsTheArchive: a producer's +// record that exists but says nothing about the archive is a verdict, not an +// absence — the bytes cannot be established, so they are not served. +func TestFetchReleaseAssetRefusesAChecksumsFileThatOmitsTheArchive(t *testing.T) { + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/repos/example-org/service-render/releases/tags/v0.1.0": + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"tag_name":"v0.1.0","assets":[{"id":12,"name":"service-render_0.1.0_linux_amd64.tar.gz","size":7,"browser_download_url":%q},{"id":13,"name":"checksums.txt","size":10,"browser_download_url":"https://github.com/example-org/service-render/releases/download/v0.1.0/checksums.txt"}]}`, testReleaseURL) + case "/repos/example-org/service-render/releases/assets/13": + fmt.Fprintf(w, "%s other.tar.gz\n", strings.Repeat("c", 64)) + default: + t.Errorf("unexpected request %s", r.URL.Path) + } + })) + t.Cleanup(api.Close) + endpoint := api.URL + "/" + client, err := github.NewClient(github.WithURLs(&endpoint, &endpoint)) + require.NoError(t, err) + _, err = FetchReleaseAsset(t.Context(), testReleaseURL, client) + require.ErrorContains(t, err, "records no digest for service-render_0.1.0_linux_amd64.tar.gz") +} + +func TestParseChecksumsReadsSha256sumOutput(t *testing.T) { + digest := strings.Repeat("A", 64) + parsed, err := parseChecksums([]byte(digest + " a.tar.gz\n" + digest + " *b.tar.gz\n\n")) + require.NoError(t, err) + require.Equal(t, map[string]string{"a.tar.gz": strings.ToLower(digest), "b.tar.gz": strings.ToLower(digest)}, parsed) + for _, raw := range []string{"abcd a.tar.gz", digest + " a.tar.gz\n" + digest + " a.tar.gz", "malformed line here", strings.Repeat("x", 70000)} { + _, err := parseChecksums([]byte(raw)) + require.Error(t, err) + } +} diff --git a/docs/agent-contract.md b/docs/agent-contract.md index 26401d0d..cf21d704 100644 --- a/docs/agent-contract.md +++ b/docs/agent-contract.md @@ -389,13 +389,24 @@ trusting their contents. Single underscores remain valid in names and release labels. Existing unambiguous cache paths are unchanged. GitHub agent downloads use `GITHUB_TOKEN`, then `GH_TOKEN`, when supplied. -Authenticated acquisition resolves the exact published tag and asset through -the GitHub API, then streams the asset through its authenticated API endpoint. -The bounded download client follows asset redirects without API credentials. -Draft releases, duplicate assets and an asset URL that disagrees with the loader -selection are refused. Without a token, the public download path is unchanged. -`manager.OpenReleaseAsset` lets a publisher verify the same selected asset with -its existing authenticated client; a private release is not a public URL probe. +`manager.GitHubReleaseToken` exposes this selection to consumers verifying +installability; a `gh` CLI login is not a loader credential. Both anonymous and +authenticated acquisition resolve the exact published tag and asset through the +GitHub API. The bounded download client follows asset redirects without API +credentials. Draft releases, duplicate assets and an asset URL that disagrees +with the loader selection are refused. An unauthenticated 404 can mean a missing +release/platform asset or a private release; supply a token with repository +access for the latter. Services do not need that download credential. + +Before extraction, downloads compare the archive's SHA-256 with its producer's +published `*checksums.txt` entry, or GitHub's asset digest when no checksum file +exists. Missing entries, malformed or duplicate checksum entries, lookup/read +failures and digest mismatches are errors. A release confirmed to offer neither +digest remains installable with an explicit unverified warning for historical +releases. Checksums provide an integrity comparison, not signed provenance. +`manager.FetchReleaseAsset` exposes the same archive and expected digest to +consumers; `OpenReleaseAsset` retains its reader/size API. Existing local and OCI +cache selection and live admission rules are unchanged. Agent downloads finish extraction before publishing a binary. Publication copies into a temporary file beside the destination, sets permissions, flushes and closes diff --git a/docs/architecture.md b/docs/architecture.md index 16730ff7..dd89df12 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -516,3 +516,15 @@ and an unrestricted one judges them all, and `Closure.Verify` takes the phase the plan is for — the closure walk follows every declared edge, so judging them all would make a build input's verdict depend on whether the walk happened to reach its consumer. + +### Signaling a detached process + +An owner persists a detached process's birth (`PID`, `BootID`, `StartID`) from +`runners/base.InspectProcess`. Before signaling, it re-inspects and compares that +birth, then supplies the current full `ProcessIdentity` to `SignalProcess`. +Process group and executable describe execution state that can change within a +birth. Core binds +kernel delivery to that birth (Linux pidfd, Darwin audit token); a missing or +replaced process returns `ErrProcessNotFound`. A PID-only record is refused. +The consumer still owns its shutdown policy and process lifecycle. Registered +process groups continue to use `TrackedProcessGroup.Signal` for group delivery. diff --git a/docs/runnable.md b/docs/runnable.md index 4abaa30d..777cfbd2 100644 --- a/docs/runnable.md +++ b/docs/runnable.md @@ -729,6 +729,30 @@ One call carries one invocation, and there is no way to hand a second to a call already in flight. `execution.concurrency` therefore bounds how many calls a facility serves at once, not a pool inside one of them. +**The request line, for either route.** A Connect procedure uses +`POST
`; a REST mutation preserves its `POST` or `PUT` verb. A prepared binding names a typed route +rather than one spelling read as both (`PreparedCall`), and a REST `GET` — the +one route core derives only as a read — carries its input as its **query** and +no body, so the method and the body are not the same for every call. +`runnable.CallRequestFor(binding, input)` is that translation, and it is the +only one: it answers the method, the absolute path, the query and the body for +whichever arm the binding sets, and refuses a verb outside `GET`/`POST`/`PUT`, +a route that is not an absolute path, and a `GET` member no query string can +spell (a null, an object, an array). The input itself must be an object: `{}` +is valid and `null` is not. Valid REST paths, including `/` and trailing slashes, +are preserved. `runnable.CallQuery` is the query half on +its own — each member one parameter of that name, a string as written, a number +or a boolean in its JSON spelling, sorted by name so one input object addresses +one request line. + +It lives here because more than one caller makes a prepared call — the task +runtime's worker dispatches an installed binding, robin's chat turn calls one +as the asking person — and the wire form of `HTTPRoute` is this package's to +define. It addresses the call and nothing more: what authorizes it, which +headers it carries, which client dials it, how an answer is classified and +where a receipt is read are the caller's, because those differ between a +durable task and a chat turn and the request line does not. + **Identity and deadline.** An invocation carries the release it invokes, an `invocation_id` for this one call, an `intent_id` stable across attempts of the caller's logical operation, and the `effect_id` an uncertain outcome is resolved diff --git a/docs/session-ledger.md b/docs/session-ledger.md index ca89b4e0..ae9893cc 100644 --- a/docs/session-ledger.md +++ b/docs/session-ledger.md @@ -9,7 +9,8 @@ Codefly already had two ownership mechanisms, and neither answers the question cleanup actually asks. - **Process groups** (`runners/base`) persist an authenticated record per spawned - group under `~/.codefly/runs/authenticated-v1/`. It is PID-reuse safe and + group under `~/.codefly/runs/authenticated-v2/` (with explicit reads of the + earlier `authenticated-v1` contract). It is PID-reuse safe and proves which groups a dead CLI left behind — but it knows nothing about containers, volumes, data, or whether a resource was ours to begin with. - **Docker containers** carried the spawning CLI's PID in a @@ -291,6 +292,48 @@ records are neither parsed nor quarantined by it, and the `v1` reader ignores decoded with unknown fields rejected, so an additive field is a schema bump, not a silent forward-compatibility trick. +## Native launch workspace + +Native tracked spawns transport the CLI's `CODEFLY_RUN_WORKSPACE` marker from +the current parent environment at spawn time. This includes materialized Nix +processes: an old cached export cannot override the current launch workspace, +and a parent without the marker clears any cached marker. Service processes +whose working directories live outside the workspace need durable attribution: +Redis, for example, can rewrite the process metadata that exposes the marker. + +New tracked spawns persist the current parent marker as `launch_workspace` in +`~/.codefly/runs/authenticated-v2/.pgid` before releasing the start gate. +The record carries the same leader/owner birth identities and group credential. +`ListProcessGroups` discovers registrations without consumer format knowledge. +`LookupProcessGroup` reads v2 and v1, refusing duplicate registrations across +namespaces. Discovery errors return no partial list and do not modify evidence. `InspectOwnership` returns `LaunchWorkspace` only when the group +authenticates; an empty value means no recorded attribution, including all v1 +records. Scope selection and refusal policy remain with the CLI. Metadata never +authorizes signaling: callers must use the existing identity-checking methods. + +The process-group registry is an explicit exception to the default generated +protobuf contract rule in [architecture](architecture.md#contract-source-of-truth). +Core retains the existing strict JSON storage contract so it can recover records +from independently released owners without rewriting their birth evidence. +The versioned namespace isolates incompatible readers. Consumers use the Core +ownership API instead of implementing a JSON decoder; this exception does not +introduce another public serialization surface or relax the protobuf default. + +The v1 wire format is unchanged. Older strict readers ignore the v2 directory; +they cannot recover its groups and must upgrade before claiming scoped cleanup +of new producers. New Core explicitly reaps both formats, locking each namespace +with its existing lock. It never copies or rewrites live v1 evidence. Unknown +namespaces remain untouched. The versioned record is the only durable provenance +source; no parallel sidecar registry or argv/environment inference is required. + +If a live recorded leader has the same process-start identity but a different +boot identity, Core reports incompatible evidence and preserves the record. +This includes older Darwin records using clock-derived boot times. It is not +proof of PID reuse or successful cleanup, and Core neither rewrites the birth +identity nor signals the unverified process. Inspection, stale recovery and +explicit signals share this refusal; an actual different start identity retains +the existing reused-process handling. + ## Startup container recovery scope The legacy PID-based startup sweep requires a `codefly.recovery-scope` @@ -317,6 +360,30 @@ environment retains its scope from first use so another flow changing the proces marker cannot redirect its cleanup. Cross-scope recovery authorizes orphan sweeping only; it never authorizes adopting another invocation's container. +Agents bind each service-owned Docker environment before first use with +`WithServiceIdentity(s.Identity)`, using the identity supplied by `Base.Load`. +The runner snapshots the exact module/service into `codefly.service`; it never +derives a parent from flattened container names, sidecar suffixes or the agent +process identity. Service version is excluded so upgrades can reuse state. +Changing the source identity afterward cannot redirect the environment, and +rebinding after first use returns an error. + +`dockerrun.ContainerServiceIdentity(labels)` reads this same attribution for +consumers such as the CLI's port guard. Missing attribution returns `nil, nil`; +a malformed present label is an error. A present conflicting label refuses +adoption and runtime-configuration replacement, including by an unbound runner. +The label alone does not authorize reuse or cleanup: consumers still check exact +recovery scope, compatible namespace, actual/logical container name and lifecycle +and publication constraints relevant to their operation. + +Legacy unlabeled containers retain the existing exact-name reuse behavior, but +remain unproven to a parent-service lookup. Core neither relabels them nor +replaces them solely to acquire attribution. In particular, a suffixed name is +not evidence that a retained container belongs to a service. Rollout requires +the creating agent to adopt this API for each container; upgrading the CLI alone +cannot attribute containers from an older agent. Recovery of old retained state +remains an explicit owner operation, never a broad cleanup. + The marker and that header are owned by `runners/recoveryscope`, which holds no Docker client: an agent validates and echoes what it inherited but never calls a daemon, so reading the marker through `runners/dockerrun` would put the Docker diff --git a/runnable/call.go b/runnable/call.go new file mode 100644 index 00000000..1df4e24a --- /dev/null +++ b/runnable/call.go @@ -0,0 +1,169 @@ +package runnable + +import ( + "encoding/json" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + + runnablev0 "github.com/codefly-dev/core/generated/go/codefly/runnable/v0" +) + +/* +How one call to a prepared binding is addressed. + +A prepared binding says which typed route its owner is reached on, and each +route spells one call differently: a Connect procedure is POSTed with the +input as its body, and a REST route is called with its own verb — POST or PUT +carrying the input as the body, GET carrying it as the query. Every caller of a +prepared binding needs that same translation, and there is more than one: the +runtime's task worker dispatches an installed binding, and robin's chat turn +calls one as the asking person. Writing it in each would put the wire form of +HTTPRoute — which this package defines and documents (docs/runnable.md, "The +REST reader") — in two places that can disagree about the same owner, so it is +one function here instead. + +It addresses the call and nothing else. What authorizes the call, what headers +it carries, which client dials it, how an answer is classified and where a +receipt is read stay with the caller: those differ between a durable task and a +chat turn, and the request line does not. +*/ + +// CallRequest is how one call to a prepared binding's operation is sent: the +// method, the absolute path, the query and the body. Path and Query are +// relative to the binding's resolved address, which a caller joins and holds +// to its own transport rule — this says nothing about where the owner is. +type CallRequest struct { + // Method is POST for a Connect procedure, and the route's own verb for a + // REST one. + Method string + // Path is the procedure or the route's path, beginning with "/". + Path string + // Query is a GET's input object spelled as its query, and empty for every + // other call. It is already encoded, and sorted by parameter name so one + // input produces one request line. + Query string + // Body is the input document a POST or PUT carries, and nil for a GET — a + // GET's input is its query, so a body beside it would be the same input + // stated twice. + Body []byte + // Connect reports a Connect procedure, which carries the protocol version + // header. A REST route is the owner's own route and carries none. + Connect bool +} + +// CallRequestFor addresses one call to the operation a prepared binding names, +// carrying input — the operation's own bounded input document, as the caller +// holds it. The binding is read as delivered: a route arm that is not set, a +// verb outside the three HTTPRoute admits, or a path that is not absolute is +// refused by name rather than dialed. +// +// It does not re-verify the binding. VerifyPrepared is what refuses a binding a +// caller must not act on, including holding a delivered GET to being a read, +// and a caller runs it once when it installs the binding rather than on every +// call. +func CallRequestFor(binding *runnablev0.PreparedBinding, input []byte) (CallRequest, error) { + switch route := binding.GetCall().GetRoute().(type) { + case *runnablev0.PreparedCall_Connect: + procedure := route.Connect.GetProcedure() + if !absolutePath(procedure) { + return CallRequest{}, fmt.Errorf("%w: prepared binding's Connect procedure %q is not an absolute path", ErrInvalid, procedure) + } + return CallRequest{Method: http.MethodPost, Path: procedure, Body: input, Connect: true}, nil + case *runnablev0.PreparedCall_Rest: + verb, path := strings.ToUpper(route.Rest.GetVerb()), route.Rest.GetPath() + if !absolutePath(path) { + return CallRequest{}, fmt.Errorf("%w: prepared binding's REST route %q is not an absolute path", ErrInvalid, path) + } + switch verb { + case http.MethodPost, http.MethodPut: + return CallRequest{Method: verb, Path: path, Body: input}, nil + case http.MethodGet: + query, err := CallQuery(input) + if err != nil { + return CallRequest{}, fmt.Errorf("%w: %s", err, Route(verb, path)) + } + return CallRequest{Method: verb, Path: path, Query: query}, nil + default: + return CallRequest{}, fmt.Errorf("%w: prepared binding's REST verb %q is none of GET, POST or PUT", ErrInvalid, verb) + } + default: + return CallRequest{}, fmt.Errorf("%w: prepared binding for operation %q names no route to call", + ErrInvalid, binding.GetOperation().GetSpelling()) + } +} + +// CallQuery spells a runnable GET's input object as its query: each member one +// parameter of that name, a string as written, a number or a boolean in its +// JSON spelling, and the empty object as the empty query. It is the encoding +// side of the rule PackageFromOpenAPIOperation derives a GET under — a query +// parameter is one scalar per name and never nullable — so a member that is +// null, an object or an array is refused rather than spelled some other way: +// the owner published no parameter a query string can carry it in, and +// inventing one would send the owner an input it never described. +// +// The parameters are sorted by name (url.Values.Encode), so one input object +// addresses one request line whatever order its members were written in. +func CallQuery(input []byte) (string, error) { + if len(input) == 0 { + return "", fmt.Errorf("%w: a GET carries its input as its query, and this call carries no input document", ErrInvalid) + } + var members map[string]json.RawMessage + if err := json.Unmarshal(input, &members); err != nil { + return "", fmt.Errorf("%w: a GET's input is one JSON object, and this call's is not: %v", ErrInvalid, err) + } + if members == nil { + return "", fmt.Errorf("%w: a GET's input must be a JSON object, not null", ErrInvalid) + } + values := url.Values{} + for name, raw := range members { + spelled, err := queryValue(raw) + if err != nil { + return "", fmt.Errorf("%w: query parameter %q %w", ErrInvalid, name, err) + } + values.Set(name, spelled) + } + return values.Encode(), nil +} + +// queryValue spells one member of a GET's input as its query value. A string +// goes in as written — url.Values escapes it — and a number or a boolean in +// the JSON spelling the document published it under, which for a number is the +// token as the caller wrote it rather than a float reformatted through a parse. +func queryValue(raw json.RawMessage) (string, error) { + trimmed := strings.TrimSpace(string(raw)) + if trimmed == "" { + return "", fmt.Errorf("carries no value") + } + switch trimmed[0] { + case '"': + var text string + if err := json.Unmarshal(raw, &text); err != nil { + return "", fmt.Errorf("is not a JSON string: %v", err) + } + return text, nil + case '{', '[': + return "", fmt.Errorf("is an object or an array, and a query carries one scalar per name") + case 'n': + return "", fmt.Errorf("is null, which a query string cannot spell") + case 't', 'f': + if trimmed != "true" && trimmed != "false" { + return "", fmt.Errorf("is not a JSON value") + } + return trimmed, nil + default: + if _, err := strconv.ParseFloat(trimmed, 64); err != nil { + return "", fmt.Errorf("is not a JSON number: %v", err) + } + return trimmed, nil + } +} + +// absolutePath holds a route to an absolute path with no query, fragment or +// space of its own: the query of a call is the input's, +// and a route carrying one would put a second input on the wire. +func absolutePath(route string) bool { + return strings.HasPrefix(route, "/") && !strings.ContainsAny(route, "?# ") +} diff --git a/runnable/call_test.go b/runnable/call_test.go new file mode 100644 index 00000000..1061283a --- /dev/null +++ b/runnable/call_test.go @@ -0,0 +1,261 @@ +package runnable_test + +import ( + "errors" + "net/http" + "net/url" + "strings" + "testing" + + basev0 "github.com/codefly-dev/core/generated/go/codefly/base/v0" + runnablev0 "github.com/codefly-dev/core/generated/go/codefly/runnable/v0" + "github.com/codefly-dev/core/runnable" +) + +func routedBinding(verb, path string) *runnablev0.PreparedBinding { + return &runnablev0.PreparedBinding{ + Operation: &runnablev0.PreparedOperation{Module: "releases", Service: "backend", Endpoint: "http", Spelling: runnable.Route(verb, path)}, + Call: &runnablev0.PreparedCall{ + Address: "http://localhost:36201", + Route: &runnablev0.PreparedCall_Rest{Rest: &runnablev0.HTTPRoute{Verb: verb, Path: path}}, + }, + } +} + +func procedureBinding(procedure string) *runnablev0.PreparedBinding { + return &runnablev0.PreparedBinding{ + Operation: &runnablev0.PreparedOperation{Module: "documents", Service: "documents", Endpoint: "connect", Spelling: procedure}, + Call: &runnablev0.PreparedCall{ + Address: "http://localhost:36202", + Route: &runnablev0.PreparedCall_Connect{Connect: &runnablev0.ConnectProcedure{Procedure: procedure}}, + }, + } +} + +// A Connect procedure is POSTed with the input as its body, and carries the +// protocol version. That is what every caller did before a REST route had a +// request line of its own, so it is pinned here first. +func TestAConnectProcedureIsPostedWithTheInputAsItsBody(t *testing.T) { + request, err := runnable.CallRequestFor(procedureBinding("/documents.v1.Documents/CountDocuments"), []byte(`{"collection":"c1"}`)) + if err != nil { + t.Fatalf("addressing a Connect call: %v", err) + } + if request.Method != http.MethodPost || request.Path != "/documents.v1.Documents/CountDocuments" { + t.Fatalf("method %q path %q", request.Method, request.Path) + } + if request.Query != "" || string(request.Body) != `{"collection":"c1"}` || !request.Connect { + t.Fatalf("query %q body %q connect %v", request.Query, request.Body, request.Connect) + } +} + +// A REST mutation keeps its body and its own verb, and is not a Connect call: +// an owner's own route carries no Connect protocol header. +func TestARestMutationKeepsItsBodyAndIsNotAConnectCall(t *testing.T) { + for _, verb := range []string{http.MethodPost, http.MethodPut} { + request, err := runnable.CallRequestFor(routedBinding(verb, "/records/apply"), []byte(`{"value":3}`)) + if err != nil { + t.Fatalf("%s: %v", verb, err) + } + if request.Method != verb || request.Path != "/records/apply" || request.Query != "" { + t.Fatalf("%s: method %q path %q query %q", verb, request.Method, request.Path, request.Query) + } + if string(request.Body) != `{"value":3}` || request.Connect { + t.Fatalf("%s: body %q connect %v", verb, request.Body, request.Connect) + } + } +} + +// A GET carries its input as its query and no body: each member one parameter +// of that name, a string as written, a number or a boolean in its JSON +// spelling. +func TestAGetCarriesItsInputAsItsQueryAndNoBody(t *testing.T) { + request, err := runnable.CallRequestFor(routedBinding(http.MethodGet, "/v1/releases/history"), + []byte(`{"repo":"obin-ai/handbook","limit":25,"ratio":1.5,"all":true}`)) + if err != nil { + t.Fatalf("addressing a GET: %v", err) + } + if request.Method != http.MethodGet || request.Path != "/v1/releases/history" { + t.Fatalf("method %q path %q", request.Method, request.Path) + } + if request.Body != nil { + t.Fatalf("a GET carries no body, and this one carries %q", request.Body) + } + values, err := url.ParseQuery(request.Query) + if err != nil { + t.Fatalf("the query does not parse: %v", err) + } + for name, want := range map[string]string{"repo": "obin-ai/handbook", "limit": "25", "ratio": "1.5", "all": "true"} { + if got := values.Get(name); got != want { + t.Fatalf("query %s=%q, want %q (whole query %q)", name, got, want, request.Query) + } + } + if len(values) != 4 { + t.Fatalf("the query carries %d parameters, want 4: %q", len(values), request.Query) + } +} + +// One input object addresses one request line: the parameters are sorted by +// name, so two spellings of the same object do not reach the owner as two +// different requests (and so an effect keyed by the call is the same effect). +func TestOneInputObjectAddressesOneRequestLine(t *testing.T) { + first, err := runnable.CallRequestFor(routedBinding(http.MethodGet, "/v1/releases/history"), []byte(`{"repo":"a","component":"b"}`)) + if err != nil { + t.Fatal(err) + } + second, err := runnable.CallRequestFor(routedBinding(http.MethodGet, "/v1/releases/history"), []byte(`{"component":"b","repo":"a"}`)) + if err != nil { + t.Fatal(err) + } + if first.Query != second.Query { + t.Fatalf("the same object addressed two request lines: %q and %q", first.Query, second.Query) + } + if first.Query != "component=b&repo=a" { + t.Fatalf("query %q", first.Query) + } +} + +// A GET with no parameter takes the empty object, and its query is empty. +func TestAGetWithNoParameterTakesTheEmptyObject(t *testing.T) { + request, err := runnable.CallRequestFor(routedBinding(http.MethodGet, "/v1/releases/recorded"), []byte(`{}`)) + if err != nil { + t.Fatalf("the empty object is a GET's input: %v", err) + } + if request.Query != "" || request.Body != nil { + t.Fatalf("query %q body %q", request.Query, request.Body) + } +} + +// A string member is escaped rather than spelled into the query raw: a value +// carrying an ampersand is one parameter, never two. +func TestAStringQueryValueIsEscapedAndStaysOneParameter(t *testing.T) { + request, err := runnable.CallRequestFor(routedBinding(http.MethodGet, "/v1/releases/history"), []byte(`{"repo":"a&limit=99"}`)) + if err != nil { + t.Fatal(err) + } + values, err := url.ParseQuery(request.Query) + if err != nil { + t.Fatalf("the query does not parse: %v", err) + } + if len(values) != 1 || values.Get("repo") != "a&limit=99" { + t.Fatalf("a value with an ampersand became %d parameters: %q", len(values), request.Query) + } +} + +// What a query string cannot spell is refused by name, never coerced: core +// derives a GET only from scalar, non-nullable query parameters, so a null, an +// object or an array means the delivered input does not match the operation's +// published contract. +func TestAGetMemberAQueryCannotSpellIsRefusedByName(t *testing.T) { + for name, input := range map[string]string{ + "null": `{"repo":null}`, + "an object": `{"repo":{"name":"handbook"}}`, + "an array": `{"repo":["a","b"]}`, + } { + _, err := runnable.CallRequestFor(routedBinding(http.MethodGet, "/v1/releases/history"), []byte(input)) + if err == nil { + t.Fatalf("%s was accepted as a query value", name) + } + if !strings.Contains(err.Error(), `"repo"`) { + t.Fatalf("%s: the refusal does not name the parameter: %v", name, err) + } + } +} + +// An input that is not one JSON object is refused: a GET's input is an object +// whose members are its parameters, and there is nothing else to spell. +func TestAGetInputThatIsNotOneObjectIsRefused(t *testing.T) { + for name, input := range map[string]string{ + "null": `null`, + "spaced null": " \n null \t", + "a number": `42`, + "a boolean": `true`, + "an array": `[1,2]`, + "a string": `"repo"`, + "nothing": ``, + "not JSON": `{`, + } { + if _, err := runnable.CallRequestFor(routedBinding(http.MethodGet, "/v1/releases/history"), []byte(input)); err == nil { + t.Fatalf("%s was accepted as a GET's input", name) + } + } +} + +// A verb outside the three HTTPRoute admits is refused by name rather than +// dialed. core's own validation refuses one on the wire; this is the caller's +// half of the same rule, for a value that reached it another way. +func TestARestVerbOutsideTheThreeIsRefusedByName(t *testing.T) { + for _, verb := range []string{http.MethodDelete, http.MethodPatch, "HEAD", "OPTIONS"} { + _, err := runnable.CallRequestFor(routedBinding(verb, "/records/apply"), []byte(`{"value":1}`)) + if err == nil { + t.Fatalf("%s was addressed as a runnable call", verb) + } + if !strings.Contains(err.Error(), verb) { + t.Fatalf("%s: the refusal does not name the verb: %v", verb, err) + } + } +} + +// A route that is not an absolute path, or that carries a query of its own, is +// refused: the query of a call is the input's, and a route carrying one would +// put a second input on the wire. +func TestARouteThatIsNotAnAbsolutePathIsRefused(t *testing.T) { + for _, path := range []string{"v1/releases", "/v1/releases?limit=1", "/v1/releases#top", "/v1 releases"} { + if _, err := runnable.CallRequestFor(routedBinding(http.MethodGet, path), []byte(`{}`)); err == nil { + t.Fatalf("route %q was addressed", path) + } + if _, err := runnable.CallRequestFor(procedureBinding(path), []byte(`{}`)); err == nil { + t.Fatalf("procedure %q was addressed", path) + } + } +} + +// A binding with no route arm reaches nothing, and says so naming the +// operation an operator is looking for. +func TestABindingWithNoRouteIsRefusedNamingItsOperation(t *testing.T) { + binding := &runnablev0.PreparedBinding{ + Operation: &runnablev0.PreparedOperation{Module: "releases", Service: "backend", Endpoint: "http", Spelling: "GET /v1/releases/history"}, + Call: &runnablev0.PreparedCall{Address: "http://localhost:36201"}, + } + _, err := runnable.CallRequestFor(binding, []byte(`{}`)) + if err == nil { + t.Fatal("a binding with no route was addressed") + } + if !strings.Contains(err.Error(), "GET /v1/releases/history") { + t.Fatalf("the refusal does not name the operation: %v", err) + } +} + +func TestCallRequestPreservesRootAndTrailingSlashRoutes(t *testing.T) { + for _, path := range []string{"/", "/v1/releases/"} { + binding := restBinding() + binding.Operation.Spelling = runnable.Route(http.MethodGet, path) + binding.Call.Route = &runnablev0.PreparedCall_Rest{Rest: &runnablev0.HTTPRoute{Verb: http.MethodGet, Path: path}} + binding.Policy.Recovery = basev0.RunnableExecution_RECOVERY_RECOMPUTE + binding.Policy.InvokeScopes[0].Actions = []string{"read"} + binding.Policy.LookupScopes = nil + decoded, err := runnable.DecodePrepared(encoded(t, binding)) + if err != nil { + t.Fatal(err) + } + request, err := runnable.CallRequestFor(decoded, []byte(`{}`)) + if err != nil { + t.Fatal(err) + } + if request.Path != path { + t.Fatalf("route %q became %q", path, request.Path) + } + } +} + +func TestCallQueryRejectsNullAsInvalid(t *testing.T) { + for _, input := range []string{"null", " \n null \t"} { + _, err := runnable.CallQuery([]byte(input)) + if !errors.Is(err, runnable.ErrInvalid) { + t.Fatalf("CallQuery(%q): %v", input, err) + } + _, err = runnable.CallRequestFor(routedBinding(http.MethodGet, "/"), []byte(input)) + if !errors.Is(err, runnable.ErrInvalid) { + t.Fatalf("CallRequestFor(%q): %v", input, err) + } + } +} diff --git a/runners/base/launch_workspace_test.go b/runners/base/launch_workspace_test.go new file mode 100644 index 00000000..e5f711ed --- /dev/null +++ b/runners/base/launch_workspace_test.go @@ -0,0 +1,39 @@ +package base_test + +import ( + "bytes" + "os" + "os/exec" + "testing" + + "github.com/codefly-dev/core/runners/base" + "github.com/stretchr/testify/require" +) + +func TestTrackedSpawnUsesCurrentLaunchWorkspace(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + for _, parent := range []string{"current workspace", "", "absent"} { + t.Run(parent, func(t *testing.T) { + t.Setenv(base.LaunchWorkspaceEnv, parent) + if parent == "absent" { + require.NoError(t, os.Unsetenv(base.LaunchWorkspaceEnv)) + } + cmd := exec.Command("sh", "-c", `printf '%s' "${CODEFLY_RUN_WORKSPACE-unset}:$SERVICE_VALUE"`) + cmd.Env = []string{base.LaunchWorkspaceEnv + "=stale cache", "SERVICE_VALUE=retained"} + original := append([]string(nil), cmd.Env...) + var output bytes.Buffer + cmd.Stdout = &output + group, err := base.StartTrackedProcessGroup(cmd) + require.NoError(t, err) + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait(); _ = group.RemoveIfDead() }) + require.NoError(t, cmd.Wait()) + require.Equal(t, original, cmd.Env, "spawn must not mutate the caller's cached environment") + expected := parent + if parent == "absent" { + expected = "unset" + } + require.Equal(t, expected+":retained", output.String()) + require.NoError(t, group.RemoveIfDead()) + }) + } +} diff --git a/runners/base/nix_materialize_integration_test.go b/runners/base/nix_materialize_integration_test.go index d661f51c..882254d8 100644 --- a/runners/base/nix_materialize_integration_test.go +++ b/runners/base/nix_materialize_integration_test.go @@ -75,6 +75,34 @@ func TestNixMaterialize_ProducesExports(t *testing.T) { assertDirectExec(t, proc) } +// Materialization may precede the run or be loaded from a previous run's +// cache. A real Nix child must carry the launching parent's current marker. +func TestNixMaterializePreservesCurrentLaunchWorkspace(t *testing.T) { + dir := requireWorkingNix(t) + t.Setenv(base.LaunchWorkspaceEnv, "materialization workspace") + env, err := base.NewNixEnvironment(t.Context(), dir) + if err != nil { + t.Fatal(err) + } + env.WithCacheDir(t.TempDir()) + if err := env.Init(t.Context()); err != nil { + t.Fatal(err) + } + t.Setenv(base.LaunchWorkspaceEnv, t.TempDir()) + proc, err := env.NewProcess("sh", "-c", `printf '%s\n' "$CODEFLY_RUN_WORKSPACE"`) + if err != nil { + t.Fatal(err) + } + out := &bufWriter{} + proc.WithOutput(out) + if err := proc.Run(t.Context()); err != nil { + t.Fatal(err) + } + if got, want := strings.TrimSpace(out.String()), os.Getenv(base.LaunchWorkspaceEnv); got != want { + t.Fatalf("child launch workspace = %q, want current parent %q", got, want) + } +} + // TestNixMaterialize_PinsGoCache asserts GOCACHE / GOMODCACHE / HOME // are set in the materialized env when the flake doesn't provide them. // Without this, `go test` loses its compile cache across every Test RPC diff --git a/runners/base/ownership.go b/runners/base/ownership.go index 0f00c765..fee094b1 100644 --- a/runners/base/ownership.go +++ b/runners/base/ownership.go @@ -6,6 +6,9 @@ import ( "fmt" "os" "path/filepath" + "strconv" + "strings" + "syscall" ) // ErrProcessGroupNotRegistered is returned when no authenticated record exists @@ -17,8 +20,10 @@ var ErrProcessNotFound = errProcessNotFound // ProcessIdentity is a process's full identity: the PID plus the boot and // start-time facts that distinguish it from a later process that happens to -// reuse the same number. Callers persisting ownership must record all of it — -// a PID on its own cannot be re-authenticated after a crash. +// reuse the same number. Persist PID, BootID and StartID to re-authenticate a +// birth after a crash; a PID alone is insufficient. PGID and Executable describe +// current execution state, which can change within that birth. Before signaling, +// re-inspect, compare the persisted birth, and supply the current full identity. type ProcessIdentity struct { PID int PGID int @@ -43,24 +48,84 @@ func InspectProcess(pid int) (ProcessIdentity, error) { }, nil } -// LookupProcessGroup returns the registered group for pgid, or +// LookupProcessGroup reads authenticated-v2 and authenticated-v1 registrations. +// It refuses ambiguous duplicate registrations. It returns the group for pgid, or // ErrProcessGroupNotRegistered. The returned handle carries the record's // authentication, so signalling through it reaches only processes that prove // membership of the group that was registered — never whatever currently // happens to hold the number. func LookupProcessGroup(pgid int) (*TrackedProcessGroup, error) { - dir, err := pgidStateDir() - if err != nil { - return nil, err - } - record, _, err := readPgidRecord(filepath.Join(dir, fmt.Sprintf("%d.pgid", pgid))) - if err != nil { + var found *TrackedProcessGroup + for _, namespace := range []string{pgidRegistryNamespace, pgidPreviousNamespace} { + dir, err := pgidNamespaceDir(namespace) + if err != nil { + return nil, err + } + record, _, err := readPgidRecord(filepath.Join(dir, fmt.Sprintf("%d.pgid", pgid))) if errors.Is(err, os.ErrNotExist) { - return nil, ErrProcessGroupNotRegistered + continue + } + if err != nil { + return nil, fmt.Errorf("read process-group record: %w", err) } - return nil, fmt.Errorf("read process-group record: %w", err) + if found != nil { + return nil, errors.New("process group has registrations in multiple namespaces") + } + found = &TrackedProcessGroup{record: record, namespace: namespace} } - return &TrackedProcessGroup{record: record}, nil + if found == nil { + return nil, ErrProcessGroupNotRegistered + } + return found, nil +} + +// ListProcessGroups returns registrations from the supported owner namespaces. +// This is discovery, not proof of liveness or authority to signal. Inspect each +// group's ownership and use its identity-checking methods for any signals. +// Unreadable, malformed or ambiguous registrations fail the enumeration; no +// partial list is returned and no records are modified. +func ListProcessGroups(ctx context.Context) ([]*TrackedProcessGroup, error) { + var groups []*TrackedProcessGroup + seen := make(map[int]bool) + for _, namespace := range []string{pgidRegistryNamespace, pgidPreviousNamespace} { + dir, err := pgidNamespaceDir(namespace) + if err != nil { + return nil, err + } + entries, err := os.ReadDir(dir) + if err != nil { + return nil, err + } + for _, entry := range entries { + if err := ctx.Err(); err != nil { + return nil, err + } + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".pgid") { + continue + } + pgid, err := strconv.Atoi(strings.TrimSuffix(entry.Name(), ".pgid")) + if err != nil || pgid <= 1 || entry.Name() != fmt.Sprintf("%d.pgid", pgid) { + return nil, fmt.Errorf("invalid process-group record name %q", entry.Name()) + } + if seen[pgid] { + continue + } + group, err := LookupProcessGroup(pgid) + if errors.Is(err, ErrProcessGroupNotRegistered) { + // Concurrent confirmed cleanup removed the registration. + continue + } + if err != nil { + return nil, err + } + seen[pgid] = true + groups = append(groups, group) + } + } + if err := ctx.Err(); err != nil { + return nil, err + } + return groups, nil } // PGID is the registered process group's identifier. @@ -92,6 +157,10 @@ type ProcessGroupOwnership struct { Authenticated bool OwnerAlive bool Members []ProcessIdentity + // LaunchWorkspace is the marker captured at spawn, available only after + // authentication. Empty means no recorded attribution (including v1). + // It supplies scope metadata, never authority to signal. + LaunchWorkspace string } // InspectOwnership uses the same membership and owner checks as registry @@ -118,6 +187,7 @@ func (group *TrackedProcessGroup) InspectOwnership(ctx context.Context) (*Proces } ownership := &ProcessGroupOwnership{Authenticated: authenticated, OwnerAlive: ownerAlive} if authenticated { + ownership.LaunchWorkspace = group.record.LaunchWorkspace for _, member := range members { ownership.Members = append(ownership.Members, ProcessIdentity{ PID: member.pid, PGID: member.pgid, BootID: member.bootID, @@ -137,3 +207,36 @@ func (group *TrackedProcessGroup) Terminate(ctx context.Context) error { } return terminateAuthenticatedGroup(ctx, group.record) } + +// SignalProcess delivers a signal only to the recorded process birth. Unlike +// TrackedProcessGroup.Signal, it addresses one process and requires no group +// registration, for owners of detached processes. The kernel handle binds +// delivery to the identity, including when the PID is reused during the call. +// An absent or replaced birth returns ErrProcessNotFound. +func SignalProcess(ctx context.Context, identity ProcessIdentity, signal syscall.Signal) error { + if err := ctx.Err(); err != nil { + return err + } + if identity.PID <= 1 || identity.PGID <= 0 || identity.BootID == "" || identity.StartID == 0 { + return fmt.Errorf("signal process: a complete process identity is required") + } + handle, err := openProcessSignalHandle(processIdentity{ + pid: identity.PID, pgid: identity.PGID, bootID: identity.BootID, + startID: identity.StartID, executable: identity.Executable, + }) + if errors.Is(err, errProcessGroupIdentityChanged) { + return ErrProcessNotFound + } + if err != nil { + return err + } + defer handle.Close() + if err := ctx.Err(); err != nil { + return err + } + if err := handle.Signal(signal); errors.Is(err, syscall.ESRCH) { + return ErrProcessNotFound + } else { + return err + } +} diff --git a/runners/base/ownership_signal_test.go b/runners/base/ownership_signal_test.go new file mode 100644 index 00000000..4a89dcdc --- /dev/null +++ b/runners/base/ownership_signal_test.go @@ -0,0 +1,52 @@ +//go:build darwin || linux + +package base + +import ( + "context" + "errors" + "path/filepath" + "syscall" + "testing" +) + +func TestSignalProcessRefusesAnotherBirthAndSignalsUnregisteredProcess(t *testing.T) { + ready := filepath.Join(t.TempDir(), "ready") + target := registryHelperCommand("member") + target.Env = append(target.Env, processGroupReadyFileEnv+"="+ready) + target.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := target.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = target.Process.Kill(); _ = target.Wait() }) + waitForTestFile(t, ready) + identity, err := InspectProcess(target.Process.Pid) + if err != nil { + t.Fatal(err) + } + stale := identity + stale.StartID++ + if err := SignalProcess(t.Context(), stale, syscall.SIGKILL); !errors.Is(err, ErrProcessNotFound) { + t.Fatalf("stale birth: %v", err) + } + if err := target.Process.Signal(syscall.Signal(0)); err != nil { + t.Fatalf("stale birth harmed target: %v", err) + } + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + if err := SignalProcess(cancelled, identity, syscall.SIGKILL); !errors.Is(err, context.Canceled) { + t.Fatalf("cancelled signal: %v", err) + } + if err := SignalProcess(t.Context(), ProcessIdentity{PID: identity.PID}, syscall.SIGKILL); err == nil { + t.Fatal("accepted PID without birth") + } + if err := SignalProcess(t.Context(), identity, syscall.SIGTERM); err != nil { + t.Fatal(err) + } + if err := target.Wait(); err != nil { + t.Fatal(err) + } + if err := SignalProcess(t.Context(), identity, syscall.SIGKILL); !errors.Is(err, ErrProcessNotFound) { + t.Fatalf("expired birth: %v", err) + } +} diff --git a/runners/base/pgid.go b/runners/base/pgid.go index e9a5de75..6ae8d4ab 100644 --- a/runners/base/pgid.go +++ b/runners/base/pgid.go @@ -12,6 +12,7 @@ import ( "os" "os/exec" "path/filepath" + "slices" "strconv" "strings" "syscall" @@ -22,6 +23,11 @@ import ( "github.com/shirou/gopsutil/v3/process" ) +// LaunchWorkspaceEnv carries the launching workspace chosen by the CLI. +// Process runners transport this marker; scope selection remains the caller's +// policy and signaling still requires authenticated process-birth evidence. +const LaunchWorkspaceEnv = "CODEFLY_RUN_WORKSPACE" + // forwardLines reads r line by line and writes each line — WITH its trailing // newline — as a single Write to w. This preserves log-prefix boundaries // (wool's per-Write prefix still applies per line) AND keeps newline @@ -74,15 +80,16 @@ func forwardLines(r io.Reader, w io.Writer) { // an upgrade, so ReapStaleProcessGroups also reconciles them — corroborating a // live leader's wall-clock start time against the record's spawn second in // place of the authentication token those records predate. Namespaced contracts -// live in subdirectories and are never scanned; only loose files matching the -// exact legacy schema are reconciled. That schema predates namespacing and was +// live in subdirectories; only explicitly supported namespaces and loose files +// matching the exact legacy schema are reconciled. That schema predates namespacing and was // shared by every agent on this runner, so a group is only ever terminated once // it is proven orphaned — a recognized record that names a live, owner-managed // group is preserved regardless of which agent wrote it. const ( pgidRootDirName = "runs" - pgidRegistryNamespace = "authenticated-v1" + pgidRegistryNamespace = "authenticated-v2" + pgidPreviousNamespace = "authenticated-v1" pgidLockName = ".reaper.lock" pgidLockRetry = 25 * time.Millisecond sigtermGrace = 3 * time.Second @@ -133,11 +140,19 @@ type recordedProcessIdentity struct { Executable string `json:"executable"` } +// pgidRecord retains the exact v1 wire shape. LaunchWorkspace is serialized +// only by the v2 envelope, never as an additive field in authenticated-v1. type pgidRecord struct { - PGID int `json:"pgid"` - Leader recordedProcessIdentity `json:"leader"` - Owner recordedProcessIdentity `json:"owner"` - Authentication string `json:"authentication"` + LaunchWorkspace string `json:"-"` + PGID int `json:"pgid"` + Leader recordedProcessIdentity `json:"leader"` + Owner recordedProcessIdentity `json:"owner"` + Authentication string `json:"authentication"` +} + +type pgidRecordV2 struct { + pgidRecord + Workspace string `json:"launch_workspace"` } type recordSnapshot struct { @@ -147,7 +162,8 @@ type recordSnapshot struct { // TrackedProcessGroup is the identity-bearing handle returned for a process // group whose private registry record was durably published. type TrackedProcessGroup struct { - record pgidRecord + namespace string + record pgidRecord } type processSignalHandle interface { @@ -175,15 +191,28 @@ func (identity processIdentity) matches(recorded recordedProcessIdentity) bool { return identity.pid == recorded.PID && identity.bootID == recorded.BootID && identity.startID == recorded.StartID } +// A changed boot representation with an unchanged start witness is not proof +// of PID reuse. Older Darwin writers used the mutable clock-derived boot time. +// Keep that evidence and refuse ownership rather than retiring a live record. +func (identity processIdentity) checkBootIdentity(recorded recordedProcessIdentity) error { + if identity.pid == recorded.PID && identity.startID == recorded.StartID && identity.bootID != recorded.BootID { + return errors.New("incompatible boot identity for unchanged process start; preserve the record and recover through its owner") + } + return nil +} + func pgidStateDir() (string, error) { + return pgidNamespaceDir(pgidRegistryNamespace) +} + +func pgidNamespaceDir(namespace string) (string, error) { home, err := os.UserHomeDir() if err != nil { return "", fmt.Errorf("cannot determine home directory: %w", err) } - // The namespace is part of the current record contract. Codefly agents are - // released independently; an agent with another contract must neither parse - // nor quarantine these records, and this reaper must never inspect theirs. - dir := filepath.Join(home, ".codefly", pgidRootDirName, pgidRegistryNamespace) + // The namespace is part of the record contract. Independently released + // readers must not parse or quarantine a format they do not support. + dir := filepath.Join(home, ".codefly", pgidRootDirName, namespace) if err := os.MkdirAll(dir, 0o700); err != nil { return "", fmt.Errorf("cannot create pgid dir: %w", err) } @@ -369,7 +398,16 @@ func startProcessGroupGate(cmd *exec.Cmd, authentication string) (*processGroupG gateFDValue := strconv.Itoa(gateFD) cmd.Path = shell cmd.Args = append([]string{"sh", "-c", "IFS= read -r codefly_gate < \"$CODEFLY_PROCESS_GROUP_GATE\" || exit 125; eval \"exec ${CODEFLY_PROCESS_GROUP_GATE_FD}<&-\"; unset CODEFLY_PROCESS_GROUP_GATE CODEFLY_PROCESS_GROUP_GATE_FD; exec \"$@\"", "codefly-process-group-gate", originalPath}, originalArgs[1:]...) - cmd.Env = append(cmd.Environ(), + // Materialized Nix environments need not inherit the parent's environment, + // and cached exports may describe a previous run. Launch attribution always + // comes from the current parent at spawn, never from those cached exports. + launchEnv := slices.DeleteFunc(cmd.Environ(), func(entry string) bool { + return strings.HasPrefix(entry, LaunchWorkspaceEnv+"=") + }) + if workspace, present := os.LookupEnv(LaunchWorkspaceEnv); present { + launchEnv = append(launchEnv, LaunchWorkspaceEnv+"="+workspace) + } + cmd.Env = append(launchEnv, groupAuthEnv+"="+authentication, "CODEFLY_PROCESS_GROUP_GATE=/dev/fd/"+gateFDValue, "CODEFLY_PROCESS_GROUP_GATE_FD="+gateFDValue) @@ -443,11 +481,12 @@ func captureProcessGroup(pgid int, authentication, executable string) (*TrackedP if leader.bootID != owner.bootID { return nil, errors.New("process identities do not belong to the current boot") } - return &TrackedProcessGroup{record: pgidRecord{ - PGID: pgid, - Leader: leader.recorded(), - Owner: owner.recorded(), - Authentication: authentication, + return &TrackedProcessGroup{namespace: pgidRegistryNamespace, record: pgidRecord{ + LaunchWorkspace: os.Getenv(LaunchWorkspaceEnv), + PGID: pgid, + Leader: leader.recorded(), + Owner: owner.recorded(), + Authentication: authentication, }}, nil } @@ -478,11 +517,14 @@ func (group *TrackedProcessGroup) persist() (returnErr error) { return fmt.Errorf("inspect process-group record: %w", err) } - content, err := json.Marshal(rec) + content, err := json.Marshal(pgidRecordV2{pgidRecord: rec, Workspace: rec.LaunchWorkspace}) if err != nil { return fmt.Errorf("encode process-group record: %w", err) } content = append(content, '\n') + if len(content) > maxRecordSize { + return errors.New("process-group record is too large") + } file, err := os.CreateTemp(dir, ".pgid-*.tmp") if err != nil { return fmt.Errorf("create process-group record: %w", err) @@ -523,7 +565,8 @@ func sameProcessGroupRegistration(first, second pgidRecord) bool { return first.PGID == second.PGID && first.Leader == second.Leader && first.Owner == second.Owner && - first.Authentication == second.Authentication + first.Authentication == second.Authentication && + first.LaunchWorkspace == second.LaunchWorkspace } // Signal delivers signal only to process instances authenticated as members @@ -640,7 +683,7 @@ func (group *TrackedProcessGroup) RemoveIfDeadContext(ctx context.Context) (retu if err := ctx.Err(); err != nil { return err } - dir, err := pgidStateDir() + dir, err := pgidNamespaceDir(group.namespace) if err != nil { return err } @@ -729,8 +772,20 @@ func readPgidRecord(path string) (pgidRecord, recordSnapshot, error) { decoder := json.NewDecoder(strings.NewReader(string(data))) decoder.DisallowUnknownFields() var rec pgidRecord - if err := decoder.Decode(&rec); err != nil { - return pgidRecord{}, snapshot, err + switch filepath.Base(filepath.Dir(path)) { + case pgidRegistryNamespace: + var envelope pgidRecordV2 + if err := decoder.Decode(&envelope); err != nil { + return pgidRecord{}, snapshot, err + } + rec = envelope.pgidRecord + rec.LaunchWorkspace = envelope.Workspace + case pgidPreviousNamespace: + if err := decoder.Decode(&rec); err != nil { + return pgidRecord{}, snapshot, err + } + default: + return pgidRecord{}, snapshot, errors.New("unsupported process-group record namespace") } if err := ensureJSONEOF(decoder); err != nil { return pgidRecord{}, snapshot, err @@ -738,6 +793,9 @@ func readPgidRecord(path string) (pgidRecord, recordSnapshot, error) { if err := rec.validate(); err != nil { return pgidRecord{}, snapshot, err } + if filepath.Base(path) != fmt.Sprintf("%d.pgid", rec.PGID) { + return pgidRecord{}, snapshot, errors.New("process-group record does not match its filename") + } return rec, snapshot, nil } @@ -837,15 +895,25 @@ func IsProcessAlive(pid int) bool { } // ReapStaleProcessGroups reconciles authenticated records in the current -// contract namespace. Live owners are preserved; groups with dead or reused -// owners are terminated. Stable malformed records are quarantined without +// and previous contract namespaces. Live owners are preserved; groups with +// dead or reused owners are terminated. Stable malformed records are quarantined without // signaling any process; records that cannot prove a stable file identity -// remain active and are reported. Codefly's own legacy plaintext records, +// remain active and are reported. Incompatible boot evidence for a live leader +// with an unchanged start identity is also preserved and reported, not retired +// as PID reuse. Codefly's own legacy plaintext records, // written loose in the registry root by an older release, are reconciled too // so their groups cannot leak across an upgrade. -func ReapStaleProcessGroups(ctx context.Context) (returnErr error) { +func ReapStaleProcessGroups(ctx context.Context) error { + var failures []error + for _, namespace := range []string{pgidRegistryNamespace, pgidPreviousNamespace} { + failures = append(failures, reapProcessGroupNamespace(ctx, namespace)) + } + return errors.Join(failures...) +} + +func reapProcessGroupNamespace(ctx context.Context, namespace string) (returnErr error) { w := wool.Get(ctx).In("base.ReapStaleProcessGroups") - dir, err := pgidStateDir() + dir, err := pgidNamespaceDir(namespace) if err != nil { return err } @@ -873,7 +941,7 @@ func ReapStaleProcessGroups(ctx context.Context) (returnErr error) { break } } - if ctx.Err() == nil { + if ctx.Err() == nil && namespace == pgidPreviousNamespace { legacyReaped, legacyErr := sweepLegacyRecords(ctx, filepath.Dir(dir)) if legacyErr != nil { failures = append(failures, legacyErr) @@ -1302,6 +1370,9 @@ func authenticateProcessGroup(ctx context.Context, rec pgidRecord) ([]processIde } for _, member := range members { if member.pid == rec.PGID { + if err := member.checkBootIdentity(rec.Leader); err != nil { + return nil, false, errors.Join(inspectErr, err) + } if member.matches(rec.Leader) { return members, true, inspectErr } @@ -1550,8 +1621,9 @@ func signalGroup(ctx context.Context, rec pgidRecord, authenticate groupAuthenti // readProcessGroupAuthentication. // // The two ways this declines to authenticate are deliberately distinct. A -// member sitting on the pgid that is not the recorded leader proves the group -// is not ours, and the caller may retire it. A readable member that lacks the +// member sitting on the pgid with a different start identity proves the group +// is not ours, and the caller may retire it. Incompatible boot evidence with +// the same start identity is undecidable and reported. A readable member that lacks the // credential proves only that we cannot tell, so it is reported rather than // retired. func authenticateOwnedProcessGroup(ctx context.Context, rec pgidRecord) ([]processIdentity, bool, error) { @@ -1561,6 +1633,9 @@ func authenticateOwnedProcessGroup(ctx context.Context, rec pgidRecord) ([]proce } for _, member := range members { if member.pid == rec.PGID { + if err := member.checkBootIdentity(rec.Leader); err != nil { + return nil, false, errors.Join(inspectErr, err) + } if member.matches(rec.Leader) { return members, true, inspectErr } diff --git a/runners/base/pgid_decision_test.go b/runners/base/pgid_decision_test.go index dc38500e..f3624e66 100644 --- a/runners/base/pgid_decision_test.go +++ b/runners/base/pgid_decision_test.go @@ -43,7 +43,7 @@ func TestProcessGroupRegistryHelper(t *testing.T) { signal.Ignore(syscall.SIGTERM) writeTestFile(t, os.Getenv(processGroupReadyFileEnv), "ready") select {} - case "owner", "stubborn-owner", "live-owner": + case "owner", "stubborn-owner", "live-owner", "workspace-owner": var stopping chan os.Signal if os.Getenv(processGroupRoleEnv) == "live-owner" { stopping = armTerminationSignal() @@ -55,6 +55,10 @@ func TestProcessGroupRegistryHelper(t *testing.T) { } leader := registryHelperCommand(leaderRole) leader.Env = append(leader.Env, processGroupReadyFileEnv+"="+os.Getenv(processGroupReadyFileEnv)) + if os.Getenv(processGroupRoleEnv) == "workspace-owner" { + // Replace the inherited environment before reaching the live child. + leader = exec.Command("env", "-i", "sh", "-c", `printf ready > "$1"; exec sleep 60`, "sh", os.Getenv(processGroupReadyFileEnv)) + } if _, err := StartTrackedProcessGroup(leader); err != nil { t.Fatal(err) } diff --git a/runners/base/process_identity_darwin.go b/runners/base/process_identity_darwin.go index c1eaf990..b080a715 100644 --- a/runners/base/process_identity_darwin.go +++ b/runners/base/process_identity_darwin.go @@ -235,8 +235,9 @@ const darwinSignalAttempts = 8 // exists and the kernel rejects the delivery with ESRCH. A caller cannot tell // that apart from the process having exited, so it moves on and a live process // survives a pass that reported no failure. The birth is therefore re-verified -// for every attempt and the version read immediately before delivery, leaving -// no room for an exec to land between the two. +// for every attempt after reading the audit-token version. A PID reused +// before the identity check is refused there; one reused afterwards cannot match that version. An +// exec between the reads or before delivery makes the kernel refuse and retry. // // Everything the exec window itself produces — a rejected token, a read that // raced the exec — is retried rather than reported, because that window is @@ -251,7 +252,7 @@ const darwinSignalAttempts = 8 func (handle *darwinProcessSignalHandle) Signal(signal syscall.Signal) error { var last error for range darwinSignalAttempts { - current, err := inspectProcess(handle.identity.pid) + unique, err := readDarwinProcessUniqueInfo(handle.identity.pid) if errors.Is(err, errProcessNotFound) { return syscall.ESRCH } @@ -259,10 +260,7 @@ func (handle *darwinProcessSignalHandle) Signal(signal syscall.Signal) error { last = err continue } - if current.pgid != handle.identity.pgid || !current.matches(handle.identity.recorded()) { - return syscall.ESRCH - } - unique, err := readDarwinProcessUniqueInfo(handle.identity.pid) + current, err := inspectProcess(handle.identity.pid) if errors.Is(err, errProcessNotFound) { return syscall.ESRCH } @@ -270,9 +268,14 @@ func (handle *darwinProcessSignalHandle) Signal(signal syscall.Signal) error { last = err continue } + if current.pgid != handle.identity.pgid || !current.matches(handle.identity.recorded()) { + return syscall.ESRCH + } err = signalProcessBirth(handle.identity.pid, unique.PIDVersion, signal) if errors.Is(err, syscall.ESRCH) { - last = err + // A stale incarnation token does not prove the birth is gone. + // Keep exhaustion actionable to both single-process and group callers. + last = fmt.Errorf("process incarnation changed during delivery") continue } return err diff --git a/runners/base/process_identity_darwin_test.go b/runners/base/process_identity_darwin_test.go index 1cdd4578..2ea0f2aa 100644 --- a/runners/base/process_identity_darwin_test.go +++ b/runners/base/process_identity_darwin_test.go @@ -1,6 +1,7 @@ package base import ( + "context" "errors" "fmt" "os" @@ -15,6 +16,59 @@ import ( "golang.org/x/sys/unix" ) +func TestDarwinLegacyBootIdentityPreservesLiveRecord(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + leader := startRegistryLeader(t, "member") + t.Cleanup(func() { stopRegistryLeader(t, leader) }) + path := recordPath(t, leader.Process.Pid) + // Control: the current boot identity and live owner preserve the record. + if err := ReapStaleProcessGroups(t.Context()); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(path); err != nil { + t.Fatal(err) + } + boot, err := unix.SysctlTimeval("kern.boottime") + if err != nil { + t.Fatal(err) + } + rewriteRecord(t, path, func(rec *pgidRecord) { + rec.Leader.BootID = fmt.Sprintf("%d:%d", boot.Sec, boot.Usec) + rec.Owner.BootID = rec.Leader.BootID + }) + before, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + group, err := LookupProcessGroup(leader.Process.Pid) + if err != nil { + t.Fatal(err) + } + for _, check := range []struct { + name string + run func() error + }{ + {"stale sweep", func() error { return ReapStaleProcessGroups(t.Context()) }}, + {"inspection", func() error { _, err := group.InspectOwnership(t.Context()); return err }}, + {"signal", func() error { return group.Signal(t.Context(), syscall.SIGTERM) }}, + {"owned authentication", func() error { + _, _, err := authenticateOwnedProcessGroup(context.Background(), group.record) + return err + }}, + } { + t.Run(check.name, func(t *testing.T) { + if err := check.run(); err == nil || !strings.Contains(err.Error(), "incompatible boot identity") { + t.Fatalf("expected explicit incompatible birth refusal, got %v", err) + } + after, err := os.ReadFile(path) + if err != nil || string(after) != string(before) { + t.Fatalf("refusal must preserve byte-identical ownership evidence: %v", err) + } + assertGroupAlive(t, leader.Process.Pid) + }) + } +} + func TestDarwinProcessNotFoundClassifiesProcessTableRaces(t *testing.T) { tests := []struct { name string diff --git a/runners/base/workspace_provenance_test.go b/runners/base/workspace_provenance_test.go new file mode 100644 index 00000000..15e7ff4b --- /dev/null +++ b/runners/base/workspace_provenance_test.go @@ -0,0 +1,164 @@ +package base + +import ( + "encoding/json" + "os" + "path/filepath" + "strconv" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestWorkspaceProvenanceSurvivesOwnerAndEnvironment(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + var groups []*TrackedProcessGroup + for _, workspace := range []string{t.TempDir(), t.TempDir()} { + t.Setenv(LaunchWorkspaceEnv, workspace) + dir := t.TempDir() + pidPath, readyPath := filepath.Join(dir, "pid"), filepath.Join(dir, "ready") + owner := registryHelperCommand("workspace-owner") + owner.Env = append(owner.Env, processGroupPIDFileEnv+"="+pidPath, processGroupReadyFileEnv+"="+readyPath) + require.NoError(t, owner.Run()) + raw, err := os.ReadFile(pidPath) + require.NoError(t, err) + pid, err := strconv.Atoi(string(raw)) + require.NoError(t, err) + t.Cleanup(func() { cleanupOrphanedGroup(pid, recordPath(t, pid)) }) + waitForTestFile(t, readyPath) + group, err := LookupProcessGroup(pid) + require.NoError(t, err) + observation, err := group.InspectOwnership(t.Context()) + require.NoError(t, err) + require.True(t, observation.Authenticated) + require.False(t, observation.OwnerAlive) + require.Equal(t, workspace, observation.LaunchWorkspace) + // env -i removed even the authentication marker; the live leader's exact + // birth remains the owner-supported proof, independent of its environment. + credential, _, err := readProcessGroupAuthentication(pid) + require.NoError(t, err) + require.Empty(t, credential) + groups = append(groups, group) + } + listed, err := ListProcessGroups(t.Context()) + require.NoError(t, err) + require.Len(t, listed, 2) + require.NoError(t, groups[0].Terminate(t.Context())) + require.NoError(t, groups[0].RemoveIfDead()) + require.False(t, groups[0].Alive()) + require.True(t, groups[1].Alive()) + sibling, err := LookupProcessGroup(groups[1].PGID()) + require.NoError(t, err) + observation, err := sibling.InspectOwnership(t.Context()) + require.NoError(t, err) + require.True(t, observation.Authenticated) + require.Equal(t, groups[1].record.LaunchWorkspace, observation.LaunchWorkspace) +} + +func TestWorkspaceProvenanceNamespaceCompatibility(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv(LaunchWorkspaceEnv, t.TempDir()) + leader := startRegistryLeader(t, "member") + defer stopRegistryLeader(t, leader) + path := recordPath(t, leader.Process.Pid) + before, err := os.ReadFile(path) + require.NoError(t, err) + previousDir, err := pgidNamespaceDir(pgidPreviousNamespace) + require.NoError(t, err) + // An old strict reader scans only v1; a valid v2 must not be quarantined. + _, err = sweepOnce(t.Context(), previousDir) + require.NoError(t, err) + after, err := os.ReadFile(path) + require.NoError(t, err) + require.Equal(t, before, after) + previousPath := filepath.Join(previousDir, filepath.Base(path)) + require.NoError(t, os.WriteFile(previousPath, before, 0600)) + _, _, err = readPgidRecord(previousPath) + require.ErrorContains(t, err, "unknown field") + require.NoError(t, os.Remove(previousPath)) + // Actual old wire bytes remain readable without manufacturing provenance. + oldBytes, err := json.Marshal(leader.group.record) + require.NoError(t, err) + require.NoError(t, os.WriteFile(previousPath, oldBytes, 0600)) + _, err = LookupProcessGroup(leader.Process.Pid) + require.ErrorContains(t, err, "multiple namespaces") + listed, err := ListProcessGroups(t.Context()) + require.ErrorContains(t, err, "multiple namespaces") + require.Nil(t, listed) + require.NoError(t, os.Remove(path)) + oldGroup, err := LookupProcessGroup(leader.Process.Pid) + require.NoError(t, err) + leader.group = oldGroup + listed, err = ListProcessGroups(t.Context()) + require.NoError(t, err) + require.Len(t, listed, 1) + observation, err := oldGroup.InspectOwnership(t.Context()) + require.NoError(t, err) + require.True(t, observation.Authenticated) + require.Empty(t, observation.LaunchWorkspace) + require.NoError(t, ReapStaleProcessGroups(t.Context())) + after, err = os.ReadFile(previousPath) + require.NoError(t, err) + require.Equal(t, oldBytes, after) +} + +func TestWorkspaceProvenanceRequiresAuthenticatedBirth(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv(LaunchWorkspaceEnv, t.TempDir()) + leader := startRegistryLeader(t, "member") + defer stopRegistryLeader(t, leader) + stale := *leader.group + stale.record.Leader.StartID++ + observation, err := stale.InspectOwnership(t.Context()) + require.NoError(t, err) + require.False(t, observation.Authenticated) + require.Empty(t, observation.LaunchWorkspace) + require.True(t, leader.group.Alive()) +} + +func TestPreviousNamespaceOrphanRecovery(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + pid, path := spawnOrphanedGroup(t, false) + t.Cleanup(func() { cleanupOrphanedGroup(pid, path) }) + rec, _, err := readPgidRecord(path) + require.NoError(t, err) + oldBytes, err := json.Marshal(rec) + require.NoError(t, err) + previousDir, err := pgidNamespaceDir(pgidPreviousNamespace) + require.NoError(t, err) + previousPath := filepath.Join(previousDir, filepath.Base(path)) + require.NoError(t, os.WriteFile(previousPath, oldBytes, 0600)) + require.NoError(t, os.Remove(path)) + require.NoError(t, ReapStaleProcessGroups(t.Context())) + require.False(t, isProcessGroupAlive(pid)) + _, err = os.Stat(previousPath) + require.ErrorIs(t, err, os.ErrNotExist) +} + +func TestProcessGroupDiscoveryRejectsMisnamedRecords(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + leader := startRegistryLeader(t, "member") + defer stopRegistryLeader(t, leader) + path := recordPath(t, leader.Process.Pid) + original, err := os.ReadFile(path) + require.NoError(t, err) + for _, name := range []string{"0" + filepath.Base(path), strconv.Itoa(leader.Process.Pid+1) + ".pgid"} { + t.Run(name, func(t *testing.T) { + wrong := filepath.Join(filepath.Dir(path), name) + require.NoError(t, os.WriteFile(wrong, original, 0600)) + defer os.Remove(wrong) + listed, err := ListProcessGroups(t.Context()) + require.Error(t, err) + require.Nil(t, listed) + if name[0] != '0' { + group, err := LookupProcessGroup(leader.Process.Pid + 1) + require.ErrorContains(t, err, "does not match its filename") + require.Nil(t, group) + } + after, err := os.ReadFile(wrong) + require.NoError(t, err) + require.Equal(t, original, after, "discovery must preserve evidence") + require.True(t, leader.group.Alive()) + }) + } +} diff --git a/runners/dockerrun/docker_runner.go b/runners/dockerrun/docker_runner.go index a80bcc71..5ff275b0 100644 --- a/runners/dockerrun/docker_runner.go +++ b/runners/dockerrun/docker_runner.go @@ -107,6 +107,8 @@ type DockerEnvironment struct { // Recovery ownership is fixed on first use. A later flow changing the // process marker must not redirect this environment's lookup or shutdown. recoveryScope *ContainerRecoveryScope + // serviceIdentity is a module/service snapshot, frozen with recoveryScope. + serviceIdentity string // carriers are the container's file-delivered configuration values, // copied into it between create and start (copyCarriers). carriers []resources.FileCarrier @@ -234,6 +236,9 @@ func (docker *DockerEnvironment) GetContainer(ctx context.Context) error { if ownershipErr := validateContainerRecoveryReuse(inspect.Config, containerConfig); ownershipErr != nil { return w.Wrapf(ownershipErr, "cannot reuse container %s; recover its owner explicitly before retrying", docker.instance.ID) } + if ownershipErr := validateContainerServiceReuse(inspect.Config.Labels, containerConfig.Labels[LabelCodeflyService]); ownershipErr != nil { + return w.Wrapf(ownershipErr, "cannot reuse container %s", docker.instance.ID) + } owned = true actualFingerprint := "" if inspect.Config != nil && inspect.Config.Labels != nil { @@ -572,6 +577,11 @@ func (docker *DockerEnvironment) createContainerConfig(ctx context.Context) *con if docker.invocation != "" { config.Labels[LabelCodeflyInvocation] = docker.invocation } + docker.mu.Lock() + if docker.serviceIdentity != "" { + config.Labels[LabelCodeflyService] = docker.serviceIdentity + } + docker.mu.Unlock() if len(docker.cmd) > 0 { w.Debug("overriding command", wool.Field("cmd", base.CommandSummary(docker.cmd))) @@ -881,6 +891,9 @@ func (docker *DockerEnvironment) IsContainerPresent(ctx context.Context) (bool, c := containers[i] for _, name := range c.Names { if name == "/"+docker.name { + if err := validateContainerServiceReuse(c.Labels, docker.serviceIdentity); err != nil { + return false, err + } if c.Labels[LabelCodeflyRecoveryScope] != scope.id || (scope.id != "" && c.Labels[LabelCodeflyOwner] != labelTrue) { return false, fmt.Errorf("container %s belongs to a different or unresolved recovery scope; refusing adoption or replacement", docker.name) diff --git a/runners/dockerrun/service_identity.go b/runners/dockerrun/service_identity.go new file mode 100644 index 00000000..f768933c --- /dev/null +++ b/runners/dockerrun/service_identity.go @@ -0,0 +1,70 @@ +package dockerrun + +import ( + "fmt" + "strings" + + "github.com/codefly-dev/core/resources" + "github.com/codefly-dev/core/resources/names" +) + +// LabelCodeflyService records the exact parent module/service. Workspace and +// host ownership are supplied separately by the recovery scope and namespace. +const LabelCodeflyService = "codefly.service" + +// WithServiceIdentity snapshots a container's parent service before first use. +// Version is deliberately excluded: a service upgrade may reuse its containers. +// An agent passes its loaded Base.Identity to each of its Docker environments. +func (docker *DockerEnvironment) WithServiceIdentity(identity *resources.ServiceIdentity) error { + key, err := containerServiceKey(identity) + if err != nil { + return err + } + docker.mu.Lock() + defer docker.mu.Unlock() + if docker.recoveryScope != nil { + return fmt.Errorf("container service identity must be bound before first use") + } + docker.serviceIdentity = key + return nil +} + +func containerServiceKey(identity *resources.ServiceIdentity) (string, error) { + if identity == nil || !names.IsModule(identity.Module) || + !names.IsPathPrefix(identity.Name) || strings.Contains(identity.Name, "/") { + return "", fmt.Errorf("container service identity requires a valid module and a single service name") + } + return identity.Unique(), nil +} + +// ContainerServiceIdentity reads exact parent attribution from Docker labels. +// A missing label returns nil, nil: it proves no parent, regardless of the +// container name. A present but malformed label returns an error. The returned +// identity contains only Module and Name; callers must separately check scope, +// namespace, concrete name, lifecycle and any resource-specific constraints. +func ContainerServiceIdentity(labels map[string]string) (*resources.ServiceIdentity, error) { + key, present := labels[LabelCodeflyService] + if !present { + return nil, nil + } + module, name, ok := strings.Cut(key, "/") + identity := &resources.ServiceIdentity{Module: module, Name: name} + if _, err := containerServiceKey(identity); !ok || err != nil { + return nil, fmt.Errorf("container has an invalid %s label", LabelCodeflyService) + } + return identity, nil +} + +func validateContainerServiceReuse(labels map[string]string, expected string) error { + identity, err := ContainerServiceIdentity(labels) + if err != nil { + return err + } + // Legacy containers remain eligible for the existing exact-name reuse + // checks. Absence is never positive parent attribution, and this does not + // relabel or replace a container merely to obtain attribution. + if identity != nil && identity.Unique() != expected { + return fmt.Errorf("container belongs to another or unbound parent service; refusing adoption or replacement") + } + return nil +} diff --git a/runners/dockerrun/service_identity_integration_test.go b/runners/dockerrun/service_identity_integration_test.go new file mode 100644 index 00000000..7c50c95d --- /dev/null +++ b/runners/dockerrun/service_identity_integration_test.go @@ -0,0 +1,104 @@ +//go:build !skip_infra + +package dockerrun + +import ( + "context" + "testing" + + "github.com/codefly-dev/core/resources" + "github.com/codefly-dev/core/runners/recoveryscope" + "github.com/docker/docker/api/types/container" + "github.com/stretchr/testify/require" +) + +func TestContainerParentServiceCreationAndReuse(t *testing.T) { + t.Setenv(recoveryscope.EnvironmentVariable, "") + scope, err := NewContainerRecoveryScope(t.TempDir(), t.TempDir(), "parent-service") + require.NoError(t, err) + require.NoError(t, SetContainerRecoveryScope(scope)) + for _, tc := range []struct { + name, existing string + legacy, changed, unbound, refused bool + }{ + {name: "same service", existing: "web/gateway"}, + {name: "same service config change", existing: "web/gateway", changed: true}, + {name: "sibling cannot adopt", existing: "web/gateway-admin", refused: true}, + {name: "sibling cannot replace", existing: "web/gateway-admin", changed: true, refused: true}, + {name: "flattened collision", existing: "web-gateway/admin", changed: true, refused: true}, + {name: "unbound cannot adopt attributed", existing: "web/gateway", unbound: true, refused: true}, + {name: "malformed present cannot adopt", refused: true}, + {name: "legacy exact reuse without attribution", legacy: true}, + } { + t.Run(tc.name, func(t *testing.T) { + name := uniqueName(t) + newEnvironment := func() *DockerEnvironment { + t.Helper() + env, err := NewDockerHeadlessEnvironment(t.Context(), resources.NewDockerImage("alpine:latest"), name) + require.NoError(t, err) + t.Cleanup(func() { _ = env.client.Close() }) + env.WithPause() + return env + } + producer := newEnvironment() + identity := &resources.ServiceIdentity{Module: "web", Name: "gateway", Version: "1.0.0"} + require.NoError(t, producer.WithServiceIdentity(identity)) + identity.Name = "mutated" + require.NoError(t, producer.Init(t.Context())) + id := producer.instance.ID + t.Cleanup(func() { _ = producer.Shutdown(context.Background()) }) + inspect, err := producer.client.ContainerInspect(t.Context(), id) + require.NoError(t, err) + require.Equal(t, "web/gateway", inspect.Config.Labels[LabelCodeflyService]) + require.Error(t, producer.WithServiceIdentity(identity), "first use freezes identity") + // Docker labels are immutable. Seed legacy/foreign declarations by + // recreating only this test's exact container, never shared state. + if tc.legacy || tc.existing != "web/gateway" { + require.NoError(t, producer.Shutdown(t.Context())) + config, host, err := producer.desiredContainerConfigs(t.Context()) + require.NoError(t, err) + config.Labels[LabelCodeflyService] = tc.existing + if tc.legacy { + delete(config.Labels, LabelCodeflyService) + } + created, err := producer.client.ContainerCreate(t.Context(), config, host, nil, nil, producer.name) + require.NoError(t, err) + id = created.ID + t.Cleanup(func() { + _ = producer.client.ContainerRemove(context.Background(), created.ID, container.RemoveOptions{Force: true}) + }) + } + reuser := newEnvironment() + t.Cleanup(func() { _ = reuser.Shutdown(context.Background()) }) + if !tc.unbound { + require.NoError(t, reuser.WithServiceIdentity(&resources.ServiceIdentity{Module: "web", Name: "gateway", Version: "2.0.0"})) + } + if tc.changed { + reuser.WithEnvironmentVariables(t.Context(), &resources.EnvironmentVariable{Key: "CONFIG_REVISION", Value: "2"}) + } + err = reuser.Init(t.Context()) + if tc.refused { + require.Error(t, err) + require.Nil(t, reuser.instance, "refusal must not acquire the foreign instance") + _, err = producer.client.ContainerInspect(t.Context(), id) + require.NoError(t, err, "refusal must not delete or replace") + return + } + require.NoError(t, err) + if tc.changed { + require.NotEqual(t, id, reuser.instance.ID) + } else { + require.Equal(t, id, reuser.instance.ID) + } + inspect, err = reuser.client.ContainerInspect(t.Context(), reuser.instance.ID) + require.NoError(t, err) + parent, err := ContainerServiceIdentity(inspect.Config.Labels) + require.NoError(t, err) + if tc.legacy { + require.Nil(t, parent, "reuse must not fabricate or stamp legacy attribution") + } else { + require.Equal(t, "web/gateway", parent.Unique()) + } + }) + } +} diff --git a/runners/dockerrun/service_identity_test.go b/runners/dockerrun/service_identity_test.go new file mode 100644 index 00000000..f97dde2b --- /dev/null +++ b/runners/dockerrun/service_identity_test.go @@ -0,0 +1,27 @@ +package dockerrun_test + +import ( + "testing" + + "github.com/codefly-dev/core/resources" + "github.com/codefly-dev/core/runners/dockerrun" + "github.com/stretchr/testify/require" +) + +func TestContainerServiceIdentityIsExact(t *testing.T) { + identity, err := dockerrun.ContainerServiceIdentity(nil) + require.NoError(t, err) + require.Nil(t, identity, "absence supplies no parent-service proof") + for _, key := range []string{"", "web", "web/gateway/extra", "web/../gateway", "/gateway", "web/", "web/gateway\n"} { + _, err := dockerrun.ContainerServiceIdentity(map[string]string{dockerrun.LabelCodeflyService: key}) + require.Error(t, err, key) + } + for _, key := range []string{"web/gateway", "web/gateway-admin", "web-gateway/admin"} { + identity, err := dockerrun.ContainerServiceIdentity(map[string]string{dockerrun.LabelCodeflyService: key}) + require.NoError(t, err) + require.Equal(t, key, identity.Unique()) + } + var env dockerrun.DockerEnvironment + require.Error(t, env.WithServiceIdentity(nil)) + require.Error(t, env.WithServiceIdentity(&resources.ServiceIdentity{Module: "web", Name: "gateway/extra"})) +} diff --git a/version/info.codefly.yaml b/version/info.codefly.yaml index e16ee137..13b55277 100644 --- a/version/info.codefly.yaml +++ b/version/info.codefly.yaml @@ -1 +1 @@ -version: 0.19.7 +version: 0.19.8