diff --git a/cmd/fluid-control/main.go b/cmd/fluid-control/main.go new file mode 100644 index 0000000..e0c2fed --- /dev/null +++ b/cmd/fluid-control/main.go @@ -0,0 +1,160 @@ +// Command fluid-control serves the FLUID control-plane APIs. +// +// It publishes revisions and records governance decisions. It is not on the +// request path: ArchitectureBlueprint.md invariant 2 requires the data plane to +// keep serving when this process is down, and that separation is only real if +// nothing here is reachable from the gateway. +// +// This surface must never be exposed to interface consumers. It is the +// mechanism that evolves the interface in response to their behaviour, and a +// consumer able to reach it could drive its own adaptation. +package main + +import ( + "context" + "crypto/ed25519" + "encoding/base64" + "errors" + "flag" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + _ "modernc.org/sqlite" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/control" + "github.com/tegwick/fluid-core/internal/evidence" + "github.com/tegwick/fluid-core/internal/intent" + "github.com/tegwick/fluid-core/internal/policy" + "github.com/tegwick/fluid-core/internal/publish" + "github.com/tegwick/fluid-core/internal/signing" +) + +func main() { + if err := run(); err != nil { + log.Fatalf("fluid-control: %v", err) + } +} + +func run() error { + var ( + addr = flag.String("addr", "127.0.0.1:8081", "listen address") + store = flag.String("store", envOr("FLUID_STORE", "fluid.db"), "evidence store path") + iface = flag.String("interface", os.Getenv("FLUID_INTERFACE"), "interface identifier") + keyID = flag.String("key-id", envOr("FLUID_SIGNING_KEY_ID", "dev"), "signing key identifier") + keyFile = flag.String("key-file", os.Getenv("FLUID_SIGNING_KEY"), "base64 ed25519 private key file") + ephemeral = flag.Bool("ephemeral-key", false, "generate a throwaway signing key (development only)") + ) + flag.Parse() + + if *iface == "" { + return errors.New("--interface is required") + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + ev, err := evidence.OpenSQLite(ctx, *store) + if err != nil { + return err + } + defer ev.Close() + + signer, err := loadSigner(*keyID, *keyFile, *ephemeral) + if err != nil { + return err + } + + intents := intent.New(ev, contract.InterfaceID(*iface)) + gate := policy.NewGate(policy.DefaultLimits()) + + pipeline, err := publish.New(publish.Options{ + Gate: gate, + Signer: signer, + Store: ev, + Intents: intents, + }) + if err != nil { + return err + } + + srv := &http.Server{ + Addr: *addr, + Handler: control.NewServer(control.NewRevisionAPI(ev, pipeline), control.NewIntentAPI(intents, gate)).Routes(), + ReadHeaderTimeout: 10 * time.Second, + } + + go func() { + <-ctx.Done() + shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = srv.Shutdown(shutdown) + }() + + log.Printf("control plane for %s listening on %s (store %s, signing key %s)", + *iface, *addr, *store, signer.KeyID()) + log.Printf("public key: %s", base64.StdEncoding.EncodeToString(signer.PublicKey())) + + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + return err + } + return nil +} + +// loadSigner resolves the signing key. +// +// An ephemeral key must be asked for explicitly. Generating one silently when +// none is configured would mean a deployment could sign revisions with a key +// nobody trusts and never notice until the router refused them. +func loadSigner(keyID, keyFile string, ephemeral bool) (*signing.Signer, error) { + if keyFile != "" { + raw, err := os.ReadFile(keyFile) + if err != nil { + return nil, fmt.Errorf("read signing key: %w", err) + } + decoded, err := base64.StdEncoding.DecodeString(trimSpace(string(raw))) + if err != nil { + return nil, fmt.Errorf("signing key is not valid base64: %w", err) + } + return signing.NewSigner(keyID, ed25519.PrivateKey(decoded)) + } + + if ephemeral { + signer, _, err := signing.GenerateKey(keyID) + if err != nil { + return nil, err + } + log.Print("WARNING: using an ephemeral signing key; revisions signed now " + + "will not verify after a restart") + return signer, nil + } + + return nil, errors.New("no signing key: pass --key-file, or --ephemeral-key for development") +} + +func trimSpace(s string) string { + start, end := 0, len(s) + for start < end && isSpace(s[start]) { + start++ + } + for end > start && isSpace(s[end-1]) { + end-- + } + return s[start:end] +} + +func isSpace(b byte) bool { + return b == ' ' || b == '\t' || b == '\n' || b == '\r' +} + +func envOr(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} diff --git a/cmd/fluid/commands.go b/cmd/fluid/commands.go index fc16d1a..41b64b5 100644 --- a/cmd/fluid/commands.go +++ b/cmd/fluid/commands.go @@ -2,6 +2,8 @@ package main import ( "context" + "crypto/ed25519" + "encoding/base64" "encoding/json" "errors" "fmt" @@ -17,6 +19,9 @@ import ( "github.com/tegwick/fluid-core/internal/contract" "github.com/tegwick/fluid-core/internal/evidence" "github.com/tegwick/fluid-core/internal/intent" + "github.com/tegwick/fluid-core/internal/policy" + "github.com/tegwick/fluid-core/internal/publish" + "github.com/tegwick/fluid-core/internal/signing" ) // open connects to the evidence store for this invocation. @@ -127,6 +132,13 @@ func runRevision(ctx context.Context, g globals, args []string) error { case "publish": fs := newFlagSet("revision publish") file := fs.String("file", "", "path to the revision descriptor (YAML or JSON)") + keyFile := fs.String("key-file", os.Getenv("FLUID_SIGNING_KEY"), "base64 ed25519 signing key file") + keyID := fs.String("key-id", envOr("FLUID_SIGNING_KEY_ID", "dev"), "signing key identifier") + ephemeral := fs.Bool("ephemeral-key", false, "sign with a throwaway key (development only)") + classes := fs.String("adaptation-classes", "", "comma-separated adaptation classes") + complexity := fs.Float64("complexity-delta", 0, "measured complexity impact") + share := fs.Float64("traffic-share", 0, "requested traffic share") + approvedBy := fs.String("approved-by", "", "authorizing operator; required by the default gate") if err := fs.Parse(args[1:]); err != nil { return err } @@ -150,43 +162,54 @@ func runRevision(ctx context.Context, g globals, args []string) error { if d.ID == "" { return errors.New("descriptor has no revision id") } - // The router refuses unsigned descriptors at runtime; refusing them here - // as well means an operator finds out at publish time rather than when - // traffic starts failing. - if d.Signature == nil { - fmt.Fprintln(os.Stderr, - "warning: descriptor is unsigned; the router will refuse it once signature enforcement is enabled") - } - body, err := json.Marshal(d) + // Publishing goes through the same pipeline the control plane uses. + // A CLI that could write a revision straight into the store would be a + // way around the deterministic policy gate, which would make the gate + // decorative for anyone with shell access. + signer, err := loadSigner(*keyID, *keyFile, *ephemeral) if err != nil { return err } - if err := store.PutRecord(ctx, contract.KindRevision, string(d.ID), body); err != nil { + + intents := intent.New(store, contract.InterfaceID(iface)) + pipeline, err := publish.New(publish.Options{ + Gate: policy.NewGate(policy.DefaultLimits()), + Signer: signer, + Store: store, + Intents: intents, + }) + if err != nil { return err } - if err := store.AppendEvent(ctx, contract.FluidEvent{ - SchemaVersion: "0.1", - ID: contract.EventID(fmt.Sprintf("EV-pub-%s-%d", d.ID, time.Now().UnixNano())), - OccurredAt: time.Now().UTC(), - EntityType: contract.FluidEventEntityTypeRevision, - EntityID: string(d.ID), - EventType: "REVISION_PUBLISHED", - Actor: contract.Actor{Type: contract.ActorTypeHuman, ID: operator()}, - Reason: fmt.Sprintf("published %s in state %s", d.ID, d.State), - }); err != nil { + var approver *contract.Actor + if *approvedBy != "" { + approver = &contract.Actor{Type: contract.ActorTypeHuman, ID: *approvedBy} + } + + candidate := publish.NewCandidate(d, contract.Actor{ + Type: contract.ActorTypeHuman, ID: operator(), + }) + + verified, report, err := pipeline.Run(ctx, candidate, publish.PromotionRequest{ + AdaptationClasses: parseClasses(*classes), + ComplexityDelta: *complexity, + RequestedTrafficShare: *share, + Approved: approver != nil, + ApprovedBy: approver, + }) + if err != nil { + printReport(report) + return err + } + if err := pipeline.Publish(ctx, verified); err != nil { return err } - // Binding the revision to its governing intent is what makes the later - // audit question answerable (Blueprint 27). - is := intent.New(store, contract.InterfaceID(iface)) - if err := is.Bind(ctx, d.ID, d.Intent.Version); err != nil { - return fmt.Errorf("published, but binding to intent %s failed: %w", d.Intent.Version, err) - } - - fmt.Printf("published %s (%s), governed by %s\n", d.ID, d.State, d.Intent.Version) + printReport(report) + fmt.Printf("\npublished %s (%s), governed by %s, signed by %s\n", + d.ID, d.State, d.Intent.Version, verified.Descriptor().Signature.KeyID) return nil case "list": @@ -508,3 +531,64 @@ func operator() string { } return "unknown" } + +// parseClasses splits a comma-separated adaptation class list. +func parseClasses(s string) []contract.AdaptationClass { + if strings.TrimSpace(s) == "" { + return nil + } + var out []contract.AdaptationClass + for _, part := range strings.Split(s, ",") { + if p := strings.TrimSpace(part); p != "" { + out = append(out, contract.AdaptationClass(p)) + } + } + return out +} + +// printReport renders the pipeline outcome stage by stage, so a rejection says +// which gate refused and why rather than only that it failed. +func printReport(r publish.Report) { + if len(r.Stages) == 0 { + return + } + w := out() + fmt.Fprintln(w, "STAGE\tRESULT\tDETAIL") + for _, s := range r.Stages { + verdict := "pass" + if !s.Passed { + verdict = "FAIL" + } + fmt.Fprintf(w, "%s\t%s\t%s\n", s.Stage, verdict, oneLine(s.Detail)) + } + _ = w.Flush() +} + +// loadSigner resolves the signing key for a publish. +// +// An ephemeral key must be requested explicitly: signing with a key nobody +// trusts produces revisions the router will refuse, and finding that out at +// publish time is far better than at traffic time. +func loadSigner(keyID, keyFile string, ephemeral bool) (*signing.Signer, error) { + if keyFile != "" { + raw, err := os.ReadFile(keyFile) + if err != nil { + return nil, fmt.Errorf("read signing key: %w", err) + } + decoded, err := base64.StdEncoding.DecodeString(strings.TrimSpace(string(raw))) + if err != nil { + return nil, fmt.Errorf("signing key is not valid base64: %w", err) + } + return signing.NewSigner(keyID, ed25519.PrivateKey(decoded)) + } + if ephemeral { + signer, _, err := signing.GenerateKey(keyID) + if err != nil { + return nil, err + } + fmt.Fprintln(os.Stderr, + "warning: signing with an ephemeral key; this revision will not verify after the key is gone") + return signer, nil + } + return nil, errors.New("no signing key: pass --key-file, or --ephemeral-key for development") +} diff --git a/cmd/fluid/flags.go b/cmd/fluid/flags.go index c137373..5c24959 100644 --- a/cmd/fluid/flags.go +++ b/cmd/fluid/flags.go @@ -20,4 +20,7 @@ func (f *flagSet) Bool(name string, value bool, usage string) *bool { return f.fs.Bool(name, value, usage) } func (f *flagSet) Int(name string, value int, usage string) *int { return f.fs.Int(name, value, usage) } -func (f *flagSet) Parse(args []string) error { return f.fs.Parse(args) } +func (f *flagSet) Float64(name string, value float64, usage string) *float64 { + return f.fs.Float64(name, value, usage) +} +func (f *flagSet) Parse(args []string) error { return f.fs.Parse(args) } diff --git a/internal/control/api.go b/internal/control/api.go new file mode 100644 index 0000000..1e83a07 --- /dev/null +++ b/internal/control/api.go @@ -0,0 +1,95 @@ +// Package control implements the FLUID control-plane APIs. +// +// ArchitectureBlueprint.md section 44 asks for a small set of internal control +// APIs. They are internal on purpose: this is the surface that publishes +// revisions and records governance decisions, and it must never be reachable by +// the consumers whose behaviour it evolves in response to. +// +// Nothing here is on the request path. The data plane keeps serving when this +// server is down (invariant 2), which is also why the CLI reads the evidence +// store directly rather than through these endpoints. +package control + +import ( + "encoding/json" + "errors" + "net/http" + "strings" + + "github.com/tegwick/fluid-core/internal/evidence" +) + +// Server exposes the control APIs over HTTP. +type Server struct { + revisions *RevisionAPI + intents *IntentAPI +} + +// NewServer wires the control APIs. +func NewServer(rev *RevisionAPI, in *IntentAPI) *Server { + return &Server{revisions: rev, intents: in} +} + +// Routes returns the control-plane mux. +func (s *Server) Routes() *http.ServeMux { + mux := http.NewServeMux() + + mux.HandleFunc("/control/v1/revisions", s.revisions.handleCollection) + mux.HandleFunc("/control/v1/revisions/", s.revisions.handleItem) + mux.HandleFunc("/control/v1/intents", s.intents.handleCollection) + mux.HandleFunc("/control/v1/intents/", s.intents.handleItem) + mux.HandleFunc("/control/v1/intents/active", s.intents.handleActive) + + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) + }) + + return mux +} + +// apiError is the control-plane error shape. +type apiError struct { + Error string `json:"error"` + Detail string `json:"detail,omitempty"` + Causes []string `json:"causes,omitempty"` +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} + +func writeError(w http.ResponseWriter, status int, msg string, causes ...string) { + writeJSON(w, status, apiError{Error: msg, Causes: causes}) +} + +// statusForStoreError maps store failures onto HTTP without leaking detail. +func statusForStoreError(err error) int { + if errors.Is(err, evidence.ErrNotFound) { + return http.StatusNotFound + } + return http.StatusInternalServerError +} + +// pathTail returns the segment after prefix, or "" when there is none. +func pathTail(path, prefix string) string { + rest := strings.TrimPrefix(path, prefix) + rest = strings.Trim(rest, "/") + if rest == "" { + return "" + } + if i := strings.Index(rest, "/"); i >= 0 { + return rest[:i] + } + return rest +} + +func decodeBody(r *http.Request, into any) error { + dec := json.NewDecoder(http.MaxBytesReader(nil, r.Body, 1<<20)) + dec.DisallowUnknownFields() + if err := dec.Decode(into); err != nil { + return err + } + return nil +} diff --git a/internal/control/api_test.go b/internal/control/api_test.go new file mode 100644 index 0000000..ca9995e --- /dev/null +++ b/internal/control/api_test.go @@ -0,0 +1,249 @@ +package control + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + + _ "modernc.org/sqlite" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" + "github.com/tegwick/fluid-core/internal/intent" + "github.com/tegwick/fluid-core/internal/policy" + "github.com/tegwick/fluid-core/internal/publish" + "github.com/tegwick/fluid-core/internal/signing" +) + +const intentDoc = `# Interface Evolution Intent + +**Current operational authority mode:** +FLUID-2 +` + +func newServer(t *testing.T) (*http.ServeMux, *evidence.SQLStore) { + t.Helper() + ctx := context.Background() + + store, err := evidence.OpenSQLite(ctx, filepath.Join(t.TempDir(), "e.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + + intents := intent.New(store, "hall-publishing") + if _, err := intents.Put(ctx, "IEI-1", intentDoc); err != nil { + t.Fatal(err) + } + if err := intents.SetActive(ctx, "IEI-1"); err != nil { + t.Fatal(err) + } + + signer, _, err := signing.GenerateKey("test-key") + if err != nil { + t.Fatal(err) + } + gate := policy.NewGate(policy.DefaultLimits()) + + pipeline, err := publish.New(publish.Options{ + Gate: gate, Signer: signer, Store: store, Intents: intents, + }) + if err != nil { + t.Fatal(err) + } + + srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate)) + return srv.Routes(), store +} + +func descriptorJSON() contract.Revision { + pc := contract.RevisionPolicyPolicyCheckPassed + return contract.Revision{ + SchemaVersion: "0.1", + ID: "R-2", + Interface: "hall-publishing", + State: contract.RevisionStateCandidate, + Contract: contract.RevisionContract{ + Type: contract.RevisionContractTypeOpenapi, + Digest: contract.Digest("sha256:" + strings.Repeat("1", 64)), + }, + Runtime: contract.RevisionRuntime{Upstream: "http://adapter:8080"}, + Intent: contract.RevisionIntent{Version: "IEI-1"}, + Policy: contract.RevisionPolicy{ + Compatibility: contract.RevisionPolicyCompatibilityAdditive, + SecurityCheck: contract.RevisionPolicySecurityCheckPassed, + PolicyCheck: &pc, + }, + } +} + +func post(t *testing.T, mux *http.ServeMux, path string, body any) *httptest.ResponseRecorder { + t.Helper() + raw, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, path, bytes.NewReader(raw))) + return rec +} + +func TestCreateRevisionVerifiesAndPublishes(t *testing.T) { + mux, store := newServer(t) + + rec := post(t, mux, "/control/v1/revisions", CreateRevisionRequest{ + Descriptor: descriptorJSON(), + Origin: contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"}, + AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassPresentation}, + ComplexityDelta: 0.2, + RequestedTrafficShare: 0.1, + Approved: true, + ApprovedBy: &contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"}, + }) + + if rec.Code != http.StatusCreated { + t.Fatalf("status = %d, body %s", rec.Code, rec.Body.String()) + } + + var resp CreateRevisionResponse + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if resp.Descriptor == nil || resp.Descriptor.Signature == nil { + t.Fatal("published descriptor came back unsigned") + } + if !resp.Report.Passed() { + t.Errorf("report says not passed: %+v", resp.Report.Stages) + } + + if _, err := store.Record(context.Background(), contract.KindRevision, "R-2"); err != nil { + t.Errorf("revision was not persisted: %v", err) + } +} + +// TestRejectedCandidateIsAResultNotAFault: a rejection is normal (invariant 14) +// and must come back with its evidence rather than as a 500. +func TestRejectedCandidateIsAResultNotAFault(t *testing.T) { + mux, _ := newServer(t) + + rec := post(t, mux, "/control/v1/revisions", CreateRevisionRequest{ + Descriptor: descriptorJSON(), + Origin: contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"}, + AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassPresentation}, + Approved: false, // the default gate requires approval + }) + + if rec.Code != http.StatusUnprocessableEntity { + t.Fatalf("status = %d, want 422; body %s", rec.Code, rec.Body.String()) + } + + var resp CreateRevisionResponse + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if resp.State != "REJECTED" { + t.Errorf("state = %q", resp.State) + } + failure, ok := resp.Report.FirstFailure() + if !ok { + t.Fatal("rejection carries no failing stage") + } + if failure.Stage != publish.StagePolicyCheck { + t.Errorf("failed at %s, want POLICY_CHECK", failure.Stage) + } + if len(failure.Evidence) == 0 { + t.Error("rejection carries no reasons") + } +} + +func TestCreateRevisionValidatesInput(t *testing.T) { + mux, _ := newServer(t) + + // A hypothesis id where a revision id belongs must not be accepted. + d := descriptorJSON() + d.ID = "H-2" + rec := post(t, mux, "/control/v1/revisions", CreateRevisionRequest{ + Descriptor: d, + Origin: contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"}, + }) + if rec.Code != http.StatusBadRequest { + t.Errorf("mis-prefixed id: status = %d, want 400", rec.Code) + } + + // Every candidate must name its origin: an artifact with no provenance + // cannot be audited later. + rec = post(t, mux, "/control/v1/revisions", CreateRevisionRequest{Descriptor: descriptorJSON()}) + if rec.Code != http.StatusBadRequest { + t.Errorf("missing origin: status = %d, want 400", rec.Code) + } +} + +func TestIntentEndpoints(t *testing.T) { + mux, _ := newServer(t) + + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/control/v1/intents/active", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("active intent: status = %d, body %s", rec.Code, rec.Body.String()) + } + + var got IntentResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got.Version != "IEI-1" || got.Mode != "FLUID-2" { + t.Errorf("active intent = %+v", got) + } + + // Historical read by version. + rec = httptest.NewRecorder() + mux.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/control/v1/intents/IEI-1", nil)) + if rec.Code != http.StatusOK { + t.Errorf("historical read: status = %d", rec.Code) + } +} + +// TestRecordedIntentCannotBeRewritten: changing what a version says would +// change what already-published revisions were governed by. +func TestRecordedIntentCannotBeRewritten(t *testing.T) { + mux, _ := newServer(t) + + rec := post(t, mux, "/control/v1/intents", RecordIntentRequest{ + Version: "IEI-1", + Document: strings.Replace(intentDoc, "FLUID-2", "FLUID-5", 1), + }) + if rec.Code != http.StatusConflict { + t.Errorf("status = %d, want 409; body %s", rec.Code, rec.Body.String()) + } +} + +func TestUnfilledTemplateIsRefused(t *testing.T) { + mux, _ := newServer(t) + + rec := post(t, mux, "/control/v1/intents", RecordIntentRequest{ + Version: "IEI-2", + Document: "mode is one of FLUID-0 FLUID-1 FLUID-2 FLUID-3 FLUID-4 FLUID-5 FLUID-6", + }) + if rec.Code != http.StatusBadRequest { + t.Errorf("an unresolved template was accepted: status = %d", rec.Code) + } +} + +func TestMethodsAreConstrained(t *testing.T) { + mux, _ := newServer(t) + for _, tc := range []struct{ method, path string }{ + {http.MethodDelete, "/control/v1/revisions/R-2"}, + {http.MethodPut, "/control/v1/intents/IEI-1"}, + } { + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, httptest.NewRequest(tc.method, tc.path, nil)) + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("%s %s: status = %d, want 405", tc.method, tc.path, rec.Code) + } + } +} diff --git a/internal/control/intent.go b/internal/control/intent.go new file mode 100644 index 0000000..f21c6b7 --- /dev/null +++ b/internal/control/intent.go @@ -0,0 +1,128 @@ +package control + +import ( + "errors" + "net/http" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/intent" + "github.com/tegwick/fluid-core/internal/policy" +) + +// IntentAPI implements ArchitectureBlueprint.md section 44.5: read active +// intent, read historical intent, validate a candidate against intent, propose +// an amendment. +// +// Amendment is deliberately a proposal and not an edit. Blueprint section 22 is +// blunt about it: the Daimon must not silently expand its own mission, and +// intent changes require a separate governance process. This API records that +// an amendment was proposed; it never enacts one. +type IntentAPI struct { + store *intent.Store + gate *policy.Gate +} + +// NewIntentAPI returns the intent API. +func NewIntentAPI(store *intent.Store, gate *policy.Gate) *IntentAPI { + return &IntentAPI{store: store, gate: gate} +} + +// IntentResponse is a recorded intent version. +type IntentResponse struct { + Version string `json:"version"` + Digest contract.Digest `json:"digest"` + Mode string `json:"mode"` + Document string `json:"document"` +} + +func (a *IntentAPI) handleCollection(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodPost: + a.record(w, r) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func (a *IntentAPI) handleActive(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + + v, err := a.store.Active(r.Context()) + if err != nil { + if errors.Is(err, intent.ErrNoActive) { + writeError(w, http.StatusNotFound, "no active interface evolution intent") + return + } + writeError(w, http.StatusInternalServerError, "could not read active intent") + return + } + writeJSON(w, http.StatusOK, toResponse(v)) +} + +func (a *IntentAPI) handleItem(w http.ResponseWriter, r *http.Request) { + version := pathTail(r.URL.Path, "/control/v1/intents") + if version == "" || version == "active" { + a.handleActive(w, r) + return + } + + switch r.Method { + case http.MethodGet: + v, err := a.store.Get(r.Context(), version) + if err != nil { + writeError(w, statusForStoreError(err), "intent version not found") + return + } + writeJSON(w, http.StatusOK, toResponse(v)) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +// RecordIntentRequest records a new intent version. +type RecordIntentRequest struct { + Version string `json:"version"` + Document string `json:"document"` + Activate bool `json:"activate,omitempty"` +} + +func (a *IntentAPI) record(w http.ResponseWriter, r *http.Request) { + var req RecordIntentRequest + if err := decodeBody(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "could not decode request", err.Error()) + return + } + + v, err := a.store.Put(r.Context(), req.Version, req.Document) + if err != nil { + if errors.Is(err, intent.ErrImmutable) { + // Rewriting a recorded version would change what already-published + // revisions were governed by, so it is a conflict, not a bad request. + writeError(w, http.StatusConflict, err.Error()) + return + } + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + if req.Activate { + if err := a.store.SetActive(r.Context(), v.Version); err != nil { + writeError(w, http.StatusInternalServerError, "recorded but not activated", err.Error()) + return + } + } + + writeJSON(w, http.StatusCreated, toResponse(v)) +} + +func toResponse(v intent.Version) IntentResponse { + return IntentResponse{ + Version: v.Version, + Digest: v.Digest, + Mode: v.Mode.String(), + Document: v.Document, + } +} diff --git a/internal/control/revision.go b/internal/control/revision.go new file mode 100644 index 0000000..5b8b8f6 --- /dev/null +++ b/internal/control/revision.go @@ -0,0 +1,176 @@ +package control + +import ( + "encoding/json" + "errors" + "net/http" + "sort" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" + "github.com/tegwick/fluid-core/internal/publish" +) + +// RevisionAPI implements ArchitectureBlueprint.md section 44.1: create, verify, +// publish, set state, query lineage. +// +// Create and verify are one operation here rather than two. A revision that +// exists but has not been verified has no use and no authority, and offering it +// as a separate resource would invite callers to treat it as one. +type RevisionAPI struct { + store evidence.Store + pipeline *publish.Pipeline +} + +// NewRevisionAPI returns the revision API. +func NewRevisionAPI(store evidence.Store, p *publish.Pipeline) *RevisionAPI { + return &RevisionAPI{store: store, pipeline: p} +} + +// CreateRevisionRequest submits a candidate for verification and publication. +type CreateRevisionRequest struct { + Descriptor contract.Revision `json:"descriptor"` + Origin contract.Actor `json:"origin"` + + AdaptationClasses []contract.AdaptationClass `json:"adaptation_classes,omitempty"` + ComplexityDelta float64 `json:"complexity_delta,omitempty"` + RequestedTrafficShare float64 `json:"requested_traffic_share,omitempty"` + Approved bool `json:"approved,omitempty"` + ApprovedBy *contract.Actor `json:"approved_by,omitempty"` +} + +// CreateRevisionResponse reports the outcome, verified or not. +type CreateRevisionResponse struct { + Revision contract.RevisionID `json:"revision"` + State string `json:"state"` + Report publish.Report `json:"report"` + // Descriptor is returned only on success, with its signature attached. + Descriptor *contract.Revision `json:"descriptor,omitempty"` +} + +func (a *RevisionAPI) handleCollection(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + a.list(w, r) + case http.MethodPost: + a.create(w, r) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func (a *RevisionAPI) handleItem(w http.ResponseWriter, r *http.Request) { + id := pathTail(r.URL.Path, "/control/v1/revisions") + if id == "" { + writeError(w, http.StatusNotFound, "no revision named") + return + } + + switch r.Method { + case http.MethodGet: + a.get(w, r, id) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +// create runs a candidate through the pipeline and publishes it if it passes. +func (a *RevisionAPI) create(w http.ResponseWriter, r *http.Request) { + var req CreateRevisionRequest + if err := decodeBody(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "could not decode request", err.Error()) + return + } + if req.Descriptor.ID == "" { + writeError(w, http.StatusBadRequest, "descriptor has no revision id") + return + } + if err := contract.RequireKind(string(req.Descriptor.ID), contract.KindRevision); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + if req.Origin.ID == "" { + writeError(w, http.StatusBadRequest, "every candidate must name its origin") + return + } + + candidate := publish.NewCandidate(req.Descriptor, req.Origin) + + verified, report, err := a.pipeline.Run(r.Context(), candidate, publish.PromotionRequest{ + AdaptationClasses: req.AdaptationClasses, + ComplexityDelta: req.ComplexityDelta, + RequestedTrafficShare: req.RequestedTrafficShare, + Approved: req.Approved, + ApprovedBy: req.ApprovedBy, + }) + if err != nil { + var rejected *publish.ErrRejected + if errors.As(err, &rejected) { + // A rejected candidate is a normal outcome (invariant 14), so it is + // reported as a result with its evidence rather than as a server + // fault. 422 says the request was well formed and the answer is no. + writeJSON(w, http.StatusUnprocessableEntity, CreateRevisionResponse{ + Revision: candidate.ID(), + State: "REJECTED", + Report: report, + }) + return + } + writeError(w, http.StatusInternalServerError, "pipeline failed", err.Error()) + return + } + + if err := a.pipeline.Publish(r.Context(), verified); err != nil { + writeError(w, http.StatusInternalServerError, "verified but not published", err.Error()) + return + } + + d := verified.Descriptor() + writeJSON(w, http.StatusCreated, CreateRevisionResponse{ + Revision: d.ID, + State: string(d.State), + Report: report, + Descriptor: &d, + }) +} + +func (a *RevisionAPI) get(w http.ResponseWriter, r *http.Request, id string) { + body, err := a.store.Record(r.Context(), contract.KindRevision, id) + if err != nil { + writeError(w, statusForStoreError(err), "revision not found") + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(body) +} + +// LineageEntry is one step in a revision's ancestry. +type LineageEntry struct { + Revision contract.RevisionID `json:"revision"` + State string `json:"state"` + Intent string `json:"intent"` +} + +func (a *RevisionAPI) list(w http.ResponseWriter, r *http.Request) { + records, err := a.store.Records(r.Context(), contract.KindRevision) + if err != nil { + writeError(w, http.StatusInternalServerError, "could not list revisions") + return + } + + out := make([]LineageEntry, 0, len(records)) + for _, body := range records { + var d contract.Revision + if err := json.Unmarshal(body, &d); err != nil { + continue + } + out = append(out, LineageEntry{ + Revision: d.ID, + State: string(d.State), + Intent: d.Intent.Version, + }) + } + sort.Slice(out, func(i, j int) bool { return out[i].Revision < out[j].Revision }) + + writeJSON(w, http.StatusOK, map[string]any{"revisions": out}) +} diff --git a/workplans/FLUID-WP-0004-publication-and-control-apis.md b/workplans/FLUID-WP-0004-publication-and-control-apis.md index da0b7bc..42505ed 100644 --- a/workplans/FLUID-WP-0004-publication-and-control-apis.md +++ b/workplans/FLUID-WP-0004-publication-and-control-apis.md @@ -4,7 +4,7 @@ type: workplan title: "Revision publication pipeline and control APIs" domain: infotech repo: fluid-core -status: active +status: done owner: worsch topic_slug: fluid-core created: "2026-09-04" @@ -26,7 +26,7 @@ bypass. ```task id: FLUID-WP-0004-T01 -status: todo +status: done priority: high state_hub_task_id: "74bf29c7-b8a6-5a18-bfc7-94641d12564b" ``` @@ -38,7 +38,7 @@ CHECK, SIGN, PUBLISH, ROUTE. Each stage emits an audit event. ```task id: FLUID-WP-0004-T02 -status: todo +status: done priority: high state_hub_task_id: "9449aa0b-3c73-56d5-8768-df94b26f0282" ``` @@ -51,7 +51,7 @@ tested. ```task id: FLUID-WP-0004-T03 -status: todo +status: done priority: high state_hub_task_id: "ba302369-3243-57ac-a34a-9a0a29002ac5" ``` @@ -62,7 +62,7 @@ Blueprint §44.1: create, verify, publish, set state, query lineage. ```task id: FLUID-WP-0004-T04 -status: todo +status: done priority: medium state_hub_task_id: "27c597da-4f58-5a0d-b70e-8ba0749d9002" ``` @@ -75,7 +75,7 @@ governed — the API records a proposal, it does not enact one. ```task id: FLUID-WP-0004-T05 -status: todo +status: done priority: high state_hub_task_id: "b2f96c6c-2c37-5892-bafd-ab9eb2c3f8c2" ``` @@ -90,7 +90,7 @@ policy. ```task id: FLUID-WP-0004-T06 -status: todo +status: done priority: medium state_hub_task_id: "70cc234b-9570-5c51-8620-1a7d32b9a25c" ```