Emit mesh/check as the bridge's own check.status, never through a tool (K1)
Review finding K1 of R4e, issue 1301 S4 (ADR 0503 §3, ADR 0513 §4: the forge requires mesh/check alone). gitea_check_status rested on "only the bridge invokes it", but node-tools invokes every tool and a tool is told no caller, so any session could set mesh/check and merge past the checker. The bridge now emits check.status on mesh.mod.mesh-delivery-forge-bridge.event. check.status, which the bus grants to its own account alone (emits), with the owed status, its key and its revision; the event id check.<key>.<rev> is the bus's dedup id. A status the bus did not take is not kept as done. OwedAct refuses mesh/check, and gitea.gitea_check_status leaves the invokes. The integration test shows the tool runner's account refused on that subject and gitea's durable consumer handed statuses emitted before it fetched.
This commit is contained in:
@@ -6,7 +6,9 @@ import (
|
||||
"fmt"
|
||||
"log"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
@@ -56,8 +58,20 @@ type performer struct {
|
||||
pub publisher
|
||||
done revisions
|
||||
logger *log.Logger
|
||||
// emit publishes the bridge's own events (check.status), and node is the machine said in their headers.
|
||||
emit emitter
|
||||
node string
|
||||
}
|
||||
|
||||
// emitter publishes a message with its headers: the bus's JetStream.
|
||||
type emitter interface {
|
||||
PublishMsg(ctx context.Context, m *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
|
||||
}
|
||||
|
||||
// checkStatusSubject is where the bridge emits check.status: under its own name, which the bus lets its account alone
|
||||
// publish (mesh-control internal/broker/nats.go, a module's own namespace).
|
||||
const checkStatusSubject = "mesh.mod." + delivery.ForgeBridgeModule + ".event." + bridge.CheckStatusEvent
|
||||
|
||||
// seen says whether the revision of the key was acted on already. A revision that cannot be read is not seen, so the
|
||||
// act is done: a status set twice is the same status.
|
||||
func (p *performer) seen(ctx context.Context, doneKey string, rev uint64) bool {
|
||||
@@ -109,12 +123,31 @@ func (p *performer) checkOwed(ctx context.Context, key string, rev uint64, value
|
||||
p.keep(ctx, doneKey, rev)
|
||||
return
|
||||
}
|
||||
if _, err := delivery.DecodeCheckerOwed(value); err != nil {
|
||||
// mesh/check is emitted on the bridge's own subject, never written through a tool (R4e finding K1): the forge module
|
||||
// consumes check.status and writes it. The event's id is the bus's dedup id, so a second emit of a revision is one.
|
||||
id, body, err := bridge.CheckStatusMessage(key, rev, value)
|
||||
if err != nil {
|
||||
p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err)
|
||||
p.keep(ctx, doneKey, rev)
|
||||
return
|
||||
}
|
||||
p.perform(ctx, doneKey, key, rev, value)
|
||||
if p.emit == nil {
|
||||
p.logger.Printf("%s at revision %d: nothing to emit %s with, so it is not kept and a restart emits it", key, rev, bridge.CheckStatusEvent)
|
||||
return
|
||||
}
|
||||
m := nats.NewMsg(checkStatusSubject)
|
||||
m.Header.Set("x-event-id", id)
|
||||
m.Header.Set("Nats-Msg-Id", id)
|
||||
m.Header.Set("x-source", delivery.ForgeBridgeModule)
|
||||
m.Header.Set("x-node", p.node)
|
||||
m.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
|
||||
m.Header.Set("content-type", "application/json")
|
||||
m.Data = body
|
||||
if _, err := p.emit.PublishMsg(ctx, m); err != nil {
|
||||
p.logger.Printf("%s at revision %d: %s could not be emitted, and is emitted again when the key is seen again: %v", key, rev, bridge.CheckStatusEvent, err)
|
||||
return
|
||||
}
|
||||
p.keep(ctx, doneKey, rev)
|
||||
}
|
||||
|
||||
// perform does the act of an owed key once for the revision, under the bookkeeping key doneKey.
|
||||
|
||||
@@ -9,6 +9,9 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
@@ -223,19 +226,50 @@ func checkerOwed(t *testing.T, number int, state, text string) (string, []byte)
|
||||
return key, v
|
||||
}
|
||||
|
||||
func TestTheCheckersStatusIsSetOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *testing.T) { // 1683/AC5 (ADR 0484 clause 2)
|
||||
// fakeEmit keeps what the performer emits on its own subjects.
|
||||
type fakeEmit struct {
|
||||
msgs []*nats.Msg
|
||||
err error
|
||||
}
|
||||
|
||||
func (e *fakeEmit) PublishMsg(_ context.Context, m *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
|
||||
if e.err != nil {
|
||||
return nil, e.err
|
||||
}
|
||||
e.msgs = append(e.msgs, m)
|
||||
return &jetstream.PubAck{}, nil
|
||||
}
|
||||
|
||||
// R4e finding K1 (issue 1301 S4): the checker's status is emitted as check.status on the bridge's own subject, never a
|
||||
// tool call, once per revision of its owed key (1683/AC5, ADR 0484 clause 2).
|
||||
func TestTheCheckersStatusIsEmittedOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *testing.T) { // K1, 1683/AC5
|
||||
f := &forgeFake{}
|
||||
p, _, done, _ := newPerformer(f)
|
||||
em := &fakeEmit{}
|
||||
p.emit, p.node = em, "n1"
|
||||
key, v := checkerOwed(t, 12, "success", "pass aaaaaaaaaaaa on bbbbbbbbbbbb")
|
||||
p.checkOwed(context.Background(), key, 3, v)
|
||||
p.checkOwed(context.Background(), key, 3, v)
|
||||
if len(f.calls) != 1 || f.calls[0].key != "gitea.gitea_check_status" ||
|
||||
f.calls[0].body != `{"context":"mesh/check","description":"pass aaaaaaaaaaaa on bbbbbbbbbbbb","owner":"novox","repo":"mesh-delivery","sha":"`+headA+`","state":"success","target_url":"https://view.example/checks/g.0123456789ab"}` {
|
||||
t.Fatalf("calls = %+v", f.calls)
|
||||
if len(f.calls) != 0 {
|
||||
t.Fatalf("a tool was called for mesh/check: %+v", f.calls)
|
||||
}
|
||||
if len(em.msgs) != 1 {
|
||||
t.Fatalf("emitted %d, want 1", len(em.msgs))
|
||||
}
|
||||
m := em.msgs[0]
|
||||
if m.Subject != "mesh.mod.mesh-delivery-forge-bridge.event.check.status" {
|
||||
t.Errorf("subject = %q", m.Subject)
|
||||
}
|
||||
if m.Header.Get("x-source") != "mesh-delivery-forge-bridge" || m.Header.Get("x-node") != "n1" || m.Header.Get("x-time") == "" ||
|
||||
m.Header.Get("content-type") != "application/json" || m.Header.Get("x-event-id") != "check."+key+".3" || m.Header.Get("Nats-Msg-Id") != "check."+key+".3" {
|
||||
t.Errorf("headers = %v", m.Header)
|
||||
}
|
||||
if want := `{"owner":"novox","repo":"mesh-delivery","sha":"` + headA + `","state":"success","description":"pass aaaaaaaaaaaa on bbbbbbbbbbbb","target_url":"https://view.example/checks/g.0123456789ab","key":"` + key + `","revision":3}`; string(m.Data) != want {
|
||||
t.Errorf("body = %s", m.Data)
|
||||
}
|
||||
p.checkOwed(context.Background(), key, 4, v)
|
||||
if len(f.calls) != 2 {
|
||||
t.Errorf("a new revision was not performed: %+v", f.calls)
|
||||
if len(em.msgs) != 2 {
|
||||
t.Errorf("a new revision was not emitted: %d", len(em.msgs))
|
||||
}
|
||||
if _, ok := done["check."+key]; !ok {
|
||||
t.Errorf("revision kept under %v, want check.%s", done, key)
|
||||
@@ -246,6 +280,22 @@ func TestTheCheckersStatusIsSetOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *tes
|
||||
}
|
||||
}
|
||||
|
||||
func TestACheckStatusTheBusDidNotTakeIsNotKeptAsDone(t *testing.T) { // K1
|
||||
p, _, done, logs := newPerformer(&forgeFake{})
|
||||
em := &fakeEmit{err: errors.New("nats: timeout")}
|
||||
p.emit = em
|
||||
key, v := checkerOwed(t, 12, "pending", "checking")
|
||||
p.checkOwed(context.Background(), key, 5, v)
|
||||
if _, ok := done["check."+key]; ok || !strings.Contains(logs.String(), "nats: timeout") {
|
||||
t.Fatalf("kept %v, log %q: a status the bus did not take is said and not kept", done, logs.String())
|
||||
}
|
||||
em.err = nil
|
||||
p.checkOwed(context.Background(), key, 5, v)
|
||||
if len(em.msgs) != 1 {
|
||||
t.Errorf("the same revision was not emitted again after the failure: %d", len(em.msgs))
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnythingButTheCheckersStatusInItsBucketIsLeft(t *testing.T) { // ADR 0484 clause 2: the checker owes one act
|
||||
f := &forgeFake{}
|
||||
p, _, done, logs := newPerformer(f)
|
||||
|
||||
+172
-32
@@ -34,6 +34,12 @@ import (
|
||||
// mesh-delivery-checker.owed.written; ADR 0484 clause 2): those entries are written by analogy with the ones the controller
|
||||
// composes for the gate's buckets and for the EVENTS consumer, not taken from a composed file.
|
||||
//
|
||||
// The user n1.node-tools is the machine's tool runner, whose account every agent session on the machine shares, carrying the
|
||||
// forge module (mesh-control internal/broker users.go KindNodeTools): it invokes every tool (node-tools' invokes "*"),
|
||||
// publishes gitea's own events and hears gitea's consumer on EVENTS, n1_node-tools, which binds the bridge's check.status
|
||||
// (gitea's consumes). Written by analogy with what the controller composes, as the checker's entries are; it is never
|
||||
// granted the bridge's namespace, which is the point of R4e finding K1.
|
||||
//
|
||||
// The lists are what mesh-controller's broker.PermissionsFor answers (internal/broker/nats.go:577) for a module principal
|
||||
// with the manifest of module.json: consumes the three gitea.pull events, invokes gitea.gitea_open_pulls,
|
||||
// gitea.gitea_commit_trees and gitea.gitea_get_file, state heads, and the stream forge; `**` is written `>` on the wire (modstreams.go
|
||||
@@ -95,10 +101,6 @@ accounts {
|
||||
"mesh.mod.gitea.call.gitea_commit_status.n1~mesh-delivery-forge-bridge"
|
||||
"mesh.mod.gitea.tool.gitea_commit_status"
|
||||
"mesh.mod.gitea.tool.gitea_commit_status.*"
|
||||
"mesh.mod.gitea.call.gitea_check_status.*.n1~mesh-delivery-forge-bridge"
|
||||
"mesh.mod.gitea.call.gitea_check_status.n1~mesh-delivery-forge-bridge"
|
||||
"mesh.mod.gitea.tool.gitea_check_status"
|
||||
"mesh.mod.gitea.tool.gitea_check_status.*"
|
||||
"mesh.mod.gitea.call.gitea_delivery_view.*.n1~mesh-delivery-forge-bridge"
|
||||
"mesh.mod.gitea.call.gitea_delivery_view.n1~mesh-delivery-forge-bridge"
|
||||
"mesh.mod.gitea.tool.gitea_delivery_view"
|
||||
@@ -112,6 +114,7 @@ accounts {
|
||||
"mesh.mod.gitea.tool.gitea_add_comment"
|
||||
"mesh.mod.gitea.tool.gitea_add_comment.*"
|
||||
"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"
|
||||
"mesh.mod.mesh-delivery-forge-bridge.event.check.status"
|
||||
"$JS.ACK.CHECKS.n1_mesh-delivery-forge-bridge.>"
|
||||
"$JS.API.CONSUMER.INFO.CHECKS.n1_mesh-delivery-forge-bridge"
|
||||
"$JS.API.CONSUMER.MSG.NEXT.CHECKS.n1_mesh-delivery-forge-bridge"
|
||||
@@ -133,6 +136,24 @@ accounts {
|
||||
allow_responses: true
|
||||
}
|
||||
}
|
||||
{ user: "n1.node-tools", password: "pw"
|
||||
permissions {
|
||||
publish { allow: [
|
||||
"mesh.mod.*.tool.>"
|
||||
"mesh.mod.*.call.>"
|
||||
"mesh.mod.gitea.event.>"
|
||||
"$JS.ACK.EVENTS.n1_node-tools.>"
|
||||
"$JS.API.CONSUMER.INFO.EVENTS.n1_node-tools"
|
||||
"$JS.API.CONSUMER.MSG.NEXT.EVENTS.n1_node-tools"
|
||||
] }
|
||||
subscribe { allow: [
|
||||
"_INBOX.n1.node-tools.>"
|
||||
"mesh.mod.gitea.tool.>"
|
||||
"mesh.mod.mesh-delivery-forge-bridge.event.check.status"
|
||||
] }
|
||||
allow_responses: true
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -732,16 +753,49 @@ func forgeMerged(t *testing.T, b *bed) []string {
|
||||
|
||||
func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] }
|
||||
|
||||
// TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey: the checker's owed bucket is made after the bridge started;
|
||||
// checkStatusAt is the subject the bridge emits check.status on, and the one gitea's consumer binds.
|
||||
const checkStatusAt = "mesh.mod.mesh-delivery-forge-bridge.event.check.status"
|
||||
|
||||
// gitesConsumer is the durable consumer of the tool runner carrying gitea on EVENTS, as the controller makes it from gitea's
|
||||
// consumes (nats.go consumerDurable <node>_<module>), filtered on the bridge's check.status.
|
||||
func giteaConsumer(ctx context.Context, t *testing.T, js jetstream.JetStream) {
|
||||
t.Helper()
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_node-tools", AckPolicy: jetstream.AckExplicitPolicy,
|
||||
FilterSubjects: []string{checkStatusAt}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// asRunner connects as the machine's tool runner, the account agent sessions share.
|
||||
func asRunner(t *testing.T, b *bed) (*nats.Conn, jetstream.JetStream) {
|
||||
t.Helper()
|
||||
nc, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo("n1.node-tools", "pw"), nats.CustomInboxPrefix("_INBOX.n1.node-tools"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
js, err := jetstream.New(nc, jetstream.WithDefaultTimeout(2*time.Second))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return nc, js
|
||||
}
|
||||
|
||||
// TestBridgeEmitsTheCheckersStatusOncePerRevisionOfItsOwedKey: the checker's owed bucket is made after the bridge started;
|
||||
// the checker writes the status mesh/check it owes on a head (the SDK's CheckerOwedKey and EncodeCheckerOwed, which the checker
|
||||
// itself writes with) and says so on its stream; the bridge sets the status on the forge within 10 s, with the context
|
||||
// mesh/check, the state and the line; the same revision is not set again, a new revision is (ADR 0484 clause 2).
|
||||
func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // 1683/AC5, ADR 0484 §2
|
||||
b := startBed(t, `{"result":{"pulls":[]}}`, nil)
|
||||
// itself writes with) and says so on its stream; the bridge emits check.status on its own subject within 10 s, with the
|
||||
// state, the line, the key and the revision; the same revision is not emitted again, a new revision is (ADR 0484 clause 2).
|
||||
// No tool is called for it (R4e finding K1). Both statuses are emitted before gitea's runner fetches anything, and its
|
||||
// durable consumer hands both over afterwards: a check.status emitted while gitea's process is not running is applied
|
||||
// when it runs.
|
||||
func TestBridgeEmitsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // K1, 1683/AC5, ADR 0484 §2
|
||||
b := startBed(t, `{"result":{"pulls":[]}}`, func(ctx context.Context, js jetstream.JetStream) { giteaConsumer(ctx, t, js) })
|
||||
calls := make(chan string, 8)
|
||||
sub, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_check_status", func(m *nats.Msg) {
|
||||
calls <- string(m.Data)
|
||||
_ = m.Respond([]byte(`{"result":{},"node":"n0"}`))
|
||||
// Every tool call is watched, and none answered here: the bed's stand-ins answer the ones the bridge asks.
|
||||
sub, err := b.ctl.Subscribe("mesh.mod.gitea.>", func(m *nats.Msg) {
|
||||
if strings.Contains(string(m.Data), "mesh/check") {
|
||||
calls <- m.Subject + " " + string(m.Data)
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -758,38 +812,124 @@ func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
put := func(state, text string) {
|
||||
events, err := b.js.Stream(b.ctx, "EVENTS")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
emitted := func() uint64 {
|
||||
info, err := events.Info(b.ctx, jetstream.WithSubjectFilter(checkStatusAt))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return info.State.Msgs
|
||||
}
|
||||
put := func(state, text string, want uint64) uint64 {
|
||||
v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := kv.Put(b.ctx, key, v); err != nil {
|
||||
rev, err := kv.Put(b.ctx, key, v)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.js.Publish(b.ctx, owedWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
next := func() string {
|
||||
select {
|
||||
case got := <-calls:
|
||||
return got
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("the status was not set within 10 s")
|
||||
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && emitted() < want; time.Sleep(50 * time.Millisecond) {
|
||||
}
|
||||
return ""
|
||||
if got := emitted(); got != want {
|
||||
t.Fatalf("EVENTS holds %d check.status within 10 s, want %d", got, want)
|
||||
}
|
||||
return rev
|
||||
}
|
||||
put("pending", "checking 0123456789ab on aaaaaaaaaaaa")
|
||||
if got := next(); !strings.Contains(got, `"context":"mesh/check"`) || !strings.Contains(got, `"state":"pending"`) || !strings.Contains(got, `"sha":"`+head+`"`) || !strings.Contains(got, `"repo":"mesh-delivery"`) {
|
||||
t.Fatalf("first call = %s", got)
|
||||
}
|
||||
put("success", "pass 0123456789ab on aaaaaaaaaaaa | unit 2 run, 3 kept")
|
||||
if got := next(); !strings.Contains(got, `"state":"success"`) || !strings.Contains(got, "unit 2 run, 3 kept") {
|
||||
t.Fatalf("second call = %s", got)
|
||||
rev1 := put("pending", "checking 0123456789ab on aaaaaaaaaaaa", 1)
|
||||
rev2 := put("success", "pass 0123456789ab on aaaaaaaaaaaa | unit 2 run, 3 kept", 2)
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
if got := emitted(); got != 2 {
|
||||
t.Errorf("EVENTS holds %d check.status, want 2: a revision was emitted twice", got)
|
||||
}
|
||||
select {
|
||||
case extra := <-calls:
|
||||
t.Errorf("a third call was made: %s", extra)
|
||||
case <-time.After(500 * time.Millisecond):
|
||||
case c := <-calls:
|
||||
t.Errorf("a tool was called with mesh/check: %s", c)
|
||||
default:
|
||||
}
|
||||
|
||||
// gitea's runner starts fetching only now, and is handed both, in order.
|
||||
_, rjs := asRunner(t, b)
|
||||
cons, err := rjs.Consumer(b.ctx, "EVENTS", "n1_node-tools")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
batch, err := cons.Fetch(2, jetstream.FetchMaxWait(5*time.Second))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got []string
|
||||
for m := range batch.Messages() {
|
||||
var c struct {
|
||||
Owner, Repo, Sha, State, Description, Key string
|
||||
Revision uint64
|
||||
}
|
||||
if err := json.Unmarshal(m.Data(), &c); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m.Headers().Get("x-source") != "mesh-delivery-forge-bridge" || m.Headers().Get("x-event-id") != fmt.Sprintf("check.%s.%d", c.Key, c.Revision) {
|
||||
t.Errorf("headers = %v", m.Headers())
|
||||
}
|
||||
got = append(got, fmt.Sprintf("%s/%s@%s %s %d %v", c.Owner, c.Repo, c.Sha[:8], c.State, c.Revision, c.Key == key))
|
||||
_ = m.Ack()
|
||||
}
|
||||
want := []string{fmt.Sprintf("novox/mesh-delivery@%s pending %d true", head[:8], rev1), fmt.Sprintf("novox/mesh-delivery@%s success %d true", head[:8], rev2)}
|
||||
if strings.Join(got, "|") != strings.Join(want, "|") {
|
||||
t.Errorf("gitea's consumer was handed %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestOnlyTheBridgesAccountMayEmitCheckStatus: the machine's tool runner, whose account every agent session there shares and
|
||||
// which invokes every tool, publishing on the bridge's check.status subject is refused by the bus, and gitea's consumer is
|
||||
// handed nothing (R4e finding K1).
|
||||
func TestOnlyTheBridgesAccountMayEmitCheckStatus(t *testing.T) { // K1
|
||||
b := startBed(t, `{"result":{"pulls":[]}}`, func(ctx context.Context, js jetstream.JetStream) { giteaConsumer(ctx, t, js) })
|
||||
refused := make(chan error, 4)
|
||||
nc, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo("n1.node-tools", "pw"), nats.CustomInboxPrefix("_INBOX.n1.node-tools"),
|
||||
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
forged, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "sha": head, "state": "success",
|
||||
"description": "by a session", "key": "x", "revision": 1 << 40})
|
||||
m := nats.NewMsg(checkStatusAt)
|
||||
m.Header.Set("x-source", "mesh-delivery-forge-bridge")
|
||||
m.Data = forged
|
||||
ctx, cancel := context.WithTimeout(b.ctx, 2*time.Second)
|
||||
defer cancel()
|
||||
if _, err := js.PublishMsg(ctx, m); err == nil {
|
||||
t.Fatal("the tool runner's account published check.status: any session could set mesh/check")
|
||||
}
|
||||
select {
|
||||
case err := <-refused:
|
||||
if !strings.Contains(err.Error(), "Permissions Violation for Publish") || !strings.Contains(err.Error(), checkStatusAt) {
|
||||
t.Errorf("refused with %v", err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Error("the bus said no permissions violation for the runner's publish")
|
||||
}
|
||||
// And the ordinary way in, a tool call, has no tool that writes mesh/check to call: the runner may call every tool, so
|
||||
// the refusal is the bus's, not a manifest's.
|
||||
info, err := b.js.Stream(b.ctx, "EVENTS")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
si, err := info.Info(b.ctx, jetstream.WithSubjectFilter(checkStatusAt))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if si.State.Msgs != 0 {
|
||||
t.Errorf("EVENTS holds %d check.status after the runner's publish, want 0", si.State.Msgs)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -201,7 +201,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err != nil {
|
||||
return fmt.Errorf("the bucket %s: %w", performedBucket, err)
|
||||
}
|
||||
perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger}
|
||||
perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger, emit: js, node: cred.Node}
|
||||
watching := make(chan error, 3)
|
||||
for _, w := range []struct {
|
||||
bucket string
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
package bridge
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
@@ -31,15 +33,60 @@ type Act struct {
|
||||
// number, body), merge.ts gitea_merge_pull_request (owner, repo, number, head, approved_heads).
|
||||
const (
|
||||
commitStatusTool = "gitea.gitea_commit_status"
|
||||
// checkStatusTool is the one tool that writes mesh/check (R4e, issue 1301's option A): gitea_commit_status refuses that
|
||||
// context, and only this bridge invokes it. Its arguments: owner, repo, sha, state, description, target_url, context.
|
||||
checkStatusTool = "gitea.gitea_check_status"
|
||||
deliveryViewTool = "gitea.gitea_delivery_view"
|
||||
noteAppendTool = "gitea.gitea_note_append"
|
||||
addCommentTool = "gitea.gitea_add_comment"
|
||||
mergePullTool = "gitea.gitea_merge_pull_request"
|
||||
)
|
||||
|
||||
// CheckStatusEvent is the event the bridge writes mesh/check with (R4e finding K1; issue 1301 S4; ADR 0503 §3): no tool
|
||||
// writes mesh/check, since a tool is told no caller and node-tools invokes every tool. The bus grants its subject,
|
||||
// mesh.mod.mesh-delivery-forge-bridge.event.check.status, to this module's own account alone (its manifest's emits), and
|
||||
// the forge module consumes it and writes the status (mesh-catalog modules/gitea check.ts).
|
||||
const CheckStatusEvent = "check.status"
|
||||
|
||||
// CheckStatus is the body of a check.status: the status the checker owes on a head, the owed key and the revision it was
|
||||
// owed at, by which the forge module drops a stale one delivered after a newer.
|
||||
type CheckStatus struct {
|
||||
Owner string `json:"owner"`
|
||||
Repo string `json:"repo"`
|
||||
Sha string `json:"sha"`
|
||||
State string `json:"state"`
|
||||
Description string `json:"description"`
|
||||
TargetURL string `json:"target_url,omitempty"`
|
||||
Key string `json:"key"`
|
||||
Revision uint64 `json:"revision"`
|
||||
}
|
||||
|
||||
var wholeSha = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`)
|
||||
|
||||
// CheckStatusMessage turns the checker's owed status (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed) at revision rev
|
||||
// into the check.status it is emitted as: its id, `check.<key>.<rev>`, the bus's dedup id too, and its body. A key that is
|
||||
// not the checker's, a value that is not its status, and a head that is not a whole sha are errors, each saying why.
|
||||
func CheckStatusMessage(key string, rev uint64, value []byte) (id string, body []byte, err error) {
|
||||
repository, _, head, err := delivery.ParseCheckerOwedKey(key)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
o, err := delivery.DecodeCheckerOwed(value)
|
||||
if err != nil {
|
||||
return "", nil, fmt.Errorf("the checker's owed status %s: %w", key, err)
|
||||
}
|
||||
if o.Act != delivery.OwedStatus || o.Context != delivery.StatusContextCheck {
|
||||
return "", nil, fmt.Errorf("the checker's owed %s is %s %s, not the status %s", key, o.Act, o.Context, delivery.StatusContextCheck)
|
||||
}
|
||||
if !wholeSha.MatchString(head) {
|
||||
return "", nil, fmt.Errorf("the checker's owed %s names the head %q, not a whole sha", key, head)
|
||||
}
|
||||
owner, repo := ownerRepo(repository)
|
||||
body, err = json.Marshal(CheckStatus{Owner: owner, Repo: repo, Sha: head, State: o.State, Description: o.Text,
|
||||
TargetURL: o.Target, Key: key, Revision: rev})
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
return fmt.Sprintf("check.%s.%d", key, rev), body, nil
|
||||
}
|
||||
|
||||
// MergeAct is the name of the act of merging a pull request.
|
||||
const MergeAct = "merge"
|
||||
|
||||
@@ -80,11 +127,11 @@ func OwedAct(key string, value []byte) (Act, error) {
|
||||
if o.Context != context {
|
||||
return Act{}, fmt.Errorf("the owed status %s says the context %q, not the %q its key names", key, o.Context, context)
|
||||
}
|
||||
tool := commitStatusTool
|
||||
if o.Context == delivery.StatusContextCheck {
|
||||
tool = checkStatusTool
|
||||
return Act{}, fmt.Errorf("the owed status %s is %s, which no tool writes: the bridge emits it as %s (CheckStatusMessage)",
|
||||
key, delivery.StatusContextCheck, CheckStatusEvent)
|
||||
}
|
||||
a.Call = Call{Tool: tool, Args: map[string]any{"owner": owner, "repo": repo, "sha": head, "context": o.Context,
|
||||
a.Call = Call{Tool: commitStatusTool, Args: map[string]any{"owner": owner, "repo": repo, "sha": head, "context": o.Context,
|
||||
"state": o.State, "description": o.Text, "target_url": o.Target}}
|
||||
case delivery.OwedView:
|
||||
if err := needsPull(); err != nil {
|
||||
|
||||
@@ -2,6 +2,7 @@ package bridge
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
@@ -43,23 +44,60 @@ func TestOwedStatusIsACommitStatusCall(t *testing.T) { // 1341/AC1
|
||||
}
|
||||
}
|
||||
|
||||
func TestOwedMeshCheckGoesThroughTheCheckStatusToolAlone(t *testing.T) { // R4e 1301-S4 option A: only gitea_check_status writes mesh/check
|
||||
// checkerKV is the status the checker owes, as it writes it (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed).
|
||||
func checkerKV(t *testing.T, o delivery.Owed) (string, []byte) {
|
||||
t.Helper()
|
||||
key, err := delivery.CheckerOwedKey("novox/mesh-delivery", 12, headA)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"})
|
||||
v, err := delivery.EncodeCheckerOwed(o)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return key, v
|
||||
}
|
||||
|
||||
// R4e finding K1 (issue 1301 S4; ADR 0503 §3): no tool writes mesh/check, since a tool is told no caller and node-tools
|
||||
// invokes every tool. The bridge emits check.status, a subject only its own account may publish, and gitea writes it.
|
||||
func TestOwedMeshCheckIsNoToolCall(t *testing.T) { // K1
|
||||
key, v := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"})
|
||||
a, err := OwedAct(key, v)
|
||||
if err == nil || !strings.Contains(err.Error(), "check.status") {
|
||||
t.Fatalf("OwedAct(mesh/check) = %+v, %v; want refused, naming check.status", a, err)
|
||||
}
|
||||
// Nor through the gate's own owed bucket: the context is refused whichever key carries it.
|
||||
gk, gv := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/check",
|
||||
delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://x.example"})
|
||||
if a, err := OwedAct(gk, gv); err == nil || !strings.Contains(err.Error(), "check.status") {
|
||||
t.Fatalf("OwedAct(gate's mesh/check) = %+v, %v; want refused", a, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckStatusMessageCarriesTheOwedStatusAndItsRevision(t *testing.T) { // K1
|
||||
key, v := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"})
|
||||
id, body, err := CheckStatusMessage(key, 42, v)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := Call{Tool: "gitea.gitea_check_status", Args: map[string]any{"owner": "novox", "repo": "mesh-delivery", "sha": headA,
|
||||
"context": "mesh/check", "state": "success", "description": "pass", "target_url": "https://view.example/checks/g.1"}}
|
||||
if !reflect.DeepEqual(a.Call, want) {
|
||||
t.Errorf("call = %+v, want %+v", a.Call, want)
|
||||
if CheckStatusEvent != "check.status" {
|
||||
t.Errorf("event = %q", CheckStatusEvent)
|
||||
}
|
||||
if id != "check."+key+".42" {
|
||||
t.Errorf("id = %q", id)
|
||||
}
|
||||
want := `{"owner":"novox","repo":"mesh-delivery","sha":"` + headA + `","state":"success","description":"pass","target_url":"https://view.example/checks/g.1","key":"` + key + `","revision":42}`
|
||||
if string(body) != want {
|
||||
t.Errorf("body = %s\nwant %s", body, want)
|
||||
}
|
||||
// A key that is not the checker's, and a value that is not its status, are no message.
|
||||
gk, gv := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/merge-gate",
|
||||
delivery.Owed{Act: delivery.OwedStatus, Context: "mesh/merge-gate", State: "success", Text: "x", Target: "https://x.example"})
|
||||
if _, _, err := CheckStatusMessage(gk, 1, gv); err == nil {
|
||||
t.Error("a key of the gate made a check.status")
|
||||
}
|
||||
if _, _, err := CheckStatusMessage(key, 1, []byte("{")); err == nil {
|
||||
t.Error("a value that is no owed status made a check.status")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -45,6 +45,7 @@ type manifest struct {
|
||||
State []state `json:"state"`
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Emits []string `json:"emits"`
|
||||
OwnSecrets map[string]string `json:"own-secrets"`
|
||||
SecretsOwner string `json:"secrets-owner"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
@@ -92,7 +93,11 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
|
||||
if want := []string{"mesh-delivery-checker.owed", "mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
if want := []string{"gitea.gitea_add_comment", "gitea.gitea_check_status", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view",
|
||||
// No tool writes mesh/check (R4e finding K1): the bridge emits check.status, which only its account may publish.
|
||||
if want := []string{"check.status"}; !reflect.DeepEqual(m.Emits, want) {
|
||||
t.Errorf("emits = %v, want %v", m.Emits, want)
|
||||
}
|
||||
if want := []string{"gitea.gitea_add_comment", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view",
|
||||
"gitea.gitea_get_file", "gitea.gitea_merge_pull_request", "gitea.gitea_note_append", "gitea.gitea_open_pulls"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
@@ -108,8 +113,8 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
|
||||
if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}}) { // 1279/AC6, 1341
|
||||
t.Errorf("state = %+v, want the buckets heads and performed", m.State)
|
||||
}
|
||||
// No other state, no bus, no seat, no tool (ADR 0437 §4).
|
||||
for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} {
|
||||
// No other state, no bus, no seat, no tool (ADR 0437 §4); its one event is check.status, asserted above (R4e finding K1).
|
||||
for _, k := range []string{"logs", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the bridge has none", k)
|
||||
}
|
||||
|
||||
@@ -20,10 +20,12 @@
|
||||
"mesh-delivery-gate.merges",
|
||||
"mesh-delivery-gate.owed"
|
||||
],
|
||||
"emits": [
|
||||
"check.status"
|
||||
],
|
||||
"invokes": [
|
||||
"gitea.gitea_merge_pull_request",
|
||||
"gitea.gitea_commit_status",
|
||||
"gitea.gitea_check_status",
|
||||
"gitea.gitea_delivery_view",
|
||||
"gitea.gitea_note_append",
|
||||
"gitea.gitea_add_comment",
|
||||
|
||||
Reference in New Issue
Block a user