design-15 S6a: secret:// operator targets + Sub-Store auto-sync

internal/server/lineusers.go

@@ -332,6 +332,9 @@ func (s *Server) handleLineUserTaskResult(r *http.Request, approval model.Approv
 			}
 		}
 	}
+	// An applied line-user change alters what nodes should serve: re-arm the
+	// Sub-Store auto-sync just like the direct mutations do (design-15 §7).
+	s.triggerVPNCoreMutation()
 	return nil
 }
 

internal/server/plugin_host.go

@@ -83,6 +83,19 @@ func (h *pluginSecretHost) Delete(_ context.Context, key string) error {
 	return h.server.store.DeletePluginSecret(bucket, entryKey)
 }
 
+// pluginSecretValue is the ONLY non-broker read path into the encrypted plugin
+// vault: in-core system code (design-15 §7 auto-sync and secret:// operator-
+// target resolution) reads through here so the store-level accessors stay
+// confined to this file (TestNoHTTPHandlerReachesThePluginSecretStore). It is
+// never exposed over HTTP and must never gain a handler.
+func (s *Server) pluginSecretValue(pluginID, key string) (string, bool) {
+	entry, ok := s.store.PluginSecret(pluginSecretBucketPrefix+pluginID, key)
+	if !ok {
+		return "", false
+	}
+	return entry.Value, true
+}
+
 // pluginTaskHost implements plugin.TaskHost (spec §9.3 step 5). The broker has already
 // checked that this invocation carries an approved operation grant and that the target
 // is one the operator approved; this side enforces everything an OPERATOR queueing the

internal/server/server.go

@@ -157,6 +157,10 @@ type Server struct {
 	// concurrent read-model builds cannot allocate two UUIDs for one line
 	// (design-15 D1).
 	lineUUIDMu sync.Mutex
+	// subStoreSync holds the debounced Sub-Store auto-sync trigger that fires
+	// after committed vpn-core mutations (design-15 §7). Nil-safe: trigger and
+	// fire paths both tolerate it.
+	subStoreSync *subStoreSyncState
 	// userLoginFail brakes FAILED password logins PER ACCOUNT (keyed on the
 	// resolved user id), mirroring the per-user 2FA limiter in intent: an attacker
 	// who already targets a known account cannot widen the password-guess budget by
@@ -363,6 +367,7 @@ func New(opts Options) (*Server, error) {
 		// consume the operator/API limiter or widen token-search throughput.
 		subLimiter:       ratelimit.New(ratelimit.Config{Rate: 2, Burst: 20}),
 		logIngestLimiter: ratelimit.New(ratelimit.Config{Rate: 5000, Burst: 10000}),
+		subStoreSync:     &subStoreSyncState{},
 		ddnsProvider: func(p model.DDNSProfile) (ddns.Provider, error) {
 			return ddns.NewProvider(p, nil)
 		},

internal/server/server_plugin_invoke.go

@@ -225,7 +225,13 @@ func (s *Server) handlePluginCall(w http.ResponseWriter, r *http.Request, p prin
 	ctx, cancel := context.WithTimeout(r.Context(), 15*time.Second)
 	defer cancel()
 	ctx = context.WithValue(ctx, pluginOperatorPrincipalKey{}, p)
-	operatorTargets, err := extractOperatorTargets(req.Payload, methodContract.OperatorTargetFields)
+	payload, err := s.resolveSecretOperatorTargets(p, req.ID, req.Payload, methodContract.OperatorTargetFields)
+	if err != nil {
+		s.recordPluginCallAudit(p, req.ID, req.Service, req.Method, scopes, "deny", err.Error())
+		writeError(w, http.StatusBadRequest, err)
+		return
+	}
+	operatorTargets, err := extractOperatorTargets(payload, methodContract.OperatorTargetFields)
 	if err != nil {
 		s.recordPluginCallAudit(p, req.ID, req.Service, req.Method, scopes, "deny", err.Error())
 		writeError(w, http.StatusBadRequest, err)
@@ -236,13 +242,13 @@ func (s *Server) handlePluginCall(w http.ResponseWriter, r *http.Request, p prin
 	err = nil
 	switch {
 	case loadedOK && loaded.Manifest.Schema == plugin.ManifestSchemaV2:
-		out, err = s.dispatchV2PluginCall(ctx, loaded, req.ID, req.Service, req.Method, req.Payload, operatorTargets)
+		out, err = s.dispatchV2PluginCall(ctx, loaded, req.ID, req.Service, req.Method, payload, operatorTargets)
 	case s.pluginRPC == nil:
 		err = errors.New("plugin rpc bus unavailable")
 	default:
-		out, err = s.pluginRPC.CallOperator(ctx, req.Service, req.Method, []byte(req.Payload))
+		out, err = s.pluginRPC.CallOperator(ctx, req.Service, req.Method, []byte(payload))
 		if errors.Is(err, plugin.ErrRPCNoService) {
-			out, err = s.callRuntimePluginService(ctx, req.ID, req.Service, req.Method, req.Payload, nil)
+			out, err = s.callRuntimePluginService(ctx, req.ID, req.Service, req.Method, payload, nil)
 		}
 	}
 	if err != nil {

internal/server/substore_sync.go

@@ -0,0 +1,158 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/LatticeNet/lattice-sdk/model"
+	"github.com/LatticeNet/lattice-server/internal/id"
+)
+
+// design-15 §7: Sub-Store deep integration without merging the plugins.
+//
+// Two pieces live in core:
+//
+//  1. secret:// operator-target resolution (handlePluginCall): a payload
+//     operator-target field may carry secret://<key>, resolved from the
+//     plugin's own encrypted secret bucket (pluginsecret:<pluginID>) before
+//     binding. The plugin's saved endpoint can therefore back http.operator.do
+//     calls without the URL ever round-tripping through the browser again.
+//  2. Auto-sync: every committed vpn-core mutation re-arms a debounced trigger
+//     (30s); when it fires and the Sub-Store companion has both a saved
+//     endpoint and autosync enabled, the server invokes its import method as
+//     the system actor with a full audit trail. There is deliberately no
+//     generic plugin event bus (design-15 appendix C).
+const (
+	subStorePluginID     = "latticenet.sub-store"
+	subStoreImportSvc    = "latticenet.sub-store/import"
+	subStoreDefaultSub   = "lattice-vpn-core"
+	subStoreAutoSyncWait = 30 * time.Second
+)
+
+// resolveSecretOperatorTargets rewrites declared payload operator-target fields
+// that carry a secret:// reference into the resolved value from the plugin's
+// encrypted secret store, returning the rewritten payload. Only declared fields
+// are touched; the resolved value never appears in audits or errors.
+func (s *Server) resolveSecretOperatorTargets(p principal, pluginID string, payload json.RawMessage, fields []string) (json.RawMessage, error) {
+	if len(fields) == 0 || len(payload) == 0 {
+		return payload, nil
+	}
+	var values map[string]json.RawMessage
+	if err := json.Unmarshal(payload, &values); err != nil {
+		return payload, nil // not an object: extractOperatorTargets reports the canonical error
+	}
+	changed := false
+	for _, field := range fields {
+		raw, ok := values[field]
+		if !ok {
+			continue
+		}
+		var ref string
+		if err := json.Unmarshal(raw, &ref); err != nil || !strings.HasPrefix(ref, "secret://") {
+			continue
+		}
+		key := strings.TrimPrefix(ref, "secret://")
+		if key == "" || len(key) > 128 || strings.ContainsAny(key, "/\x00") {
+			return nil, fmt.Errorf("operator target field %q has an invalid secret reference", field)
+		}
+		value, ok := s.pluginSecretValue(pluginID, key)
+		if !ok || strings.TrimSpace(value) == "" {
+			return nil, fmt.Errorf("operator target field %q references a secret that is not saved; save the endpoint first", field)
+		}
+		values[field] = json.RawMessage(strconv.Quote(strings.TrimSpace(value)))
+		changed = true
+		s.recordPrincipalAudit(p, model.AuditEvent{
+			ID: id.New("audit"), Action: "plugin.operator_target.secret_resolve", Scope: "proxy:read",
+			Metadata: map[string]string{"plugin_id": pluginID, "field": field, "key": key},
+		})
+	}
+	if !changed {
+		return payload, nil
+	}
+	return json.Marshal(values)
+}
+
+// ── auto-sync on vpn-core mutations ───────────────────────────────────────────
+
+// subStoreSyncState holds the debounced auto-sync trigger state. invoke is a
+// test seam; production leaves it nil and falls back to callRuntimePluginService.
+type subStoreSyncState struct {
+	mu     sync.Mutex
+	timer  *time.Timer
+	invoke func(ctx context.Context, pluginID, service, method string, payload json.RawMessage, operatorTargets []string) ([]byte, error)
+}
+
+// triggerVPNCoreMutation re-arms the debounced auto-sync. It is called after
+// every committed vpn-core write (identity CRUD, bindings, rotation, and
+// applied line-user changes) and never blocks the write path.
+func (s *Server) triggerVPNCoreMutation() {
+	if s.subStoreSync == nil {
+		return
+	}
+	s.subStoreSync.mu.Lock()
+	defer s.subStoreSync.mu.Unlock()
+	if s.subStoreSync.timer != nil {
+		s.subStoreSync.timer.Stop()
+	}
+	s.subStoreSync.timer = time.AfterFunc(subStoreAutoSyncWait, func() {
+		if err := s.runSubStoreAutoSync(); err != nil {
+			s.logger.Printf("sub-store autosync: %v", err)
+		}
+	})
+}
+
+// subStoreAutoSyncTarget reads the companion's saved endpoint + autosync flag
+// from its encrypted secret namespace. (endpoint, true) only when both exist.
+func (s *Server) subStoreAutoSyncTarget() (string, bool) {
+	endpoint, ok := s.pluginSecretValue(subStorePluginID, "endpoint")
+	if !ok || strings.TrimSpace(endpoint) == "" {
+		return "", false
+	}
+	flag, ok := s.pluginSecretValue(subStorePluginID, "autosync")
+	if !ok || strings.TrimSpace(flag) != "1" {
+		return "", false
+	}
+	return strings.TrimSpace(endpoint), true
+}
+
+// runSubStoreAutoSync performs one debounced sync. Skipping (no saved endpoint,
+// autosync off, plugin inactive, no runtime) is silent; invoking is audited
+// with the system actor, and a failed import surfaces as a deny audit — never
+// a retry storm.
+func (s *Server) runSubStoreAutoSync() error {
+	if s.subStoreSync == nil || s.pluginRuntime == nil {
+		return nil
+	}
+	endpoint, enabled := s.subStoreAutoSyncTarget()
+	if !enabled || !s.pluginIsActive(subStorePluginID) {
+		return nil
+	}
+	payload, err := json.Marshal(map[string]string{"base_url": endpoint, "sub_name": subStoreDefaultSub})
+	if err != nil {
+		return err
+	}
+	audit := model.AuditEvent{
+		ID: id.New("audit"), At: s.now(), Action: "substore.autosync", Scope: "proxy:admin", ActorID: "system",
+		Metadata: map[string]string{"plugin_id": subStorePluginID, "method": "import"},
+	}
+	invoke := s.subStoreSync.invoke
+	if invoke == nil {
+		invoke = s.callRuntimePluginService
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	if _, err := invoke(ctx, subStorePluginID, subStoreImportSvc, "import", payload, []string{endpoint}); err != nil {
+		audit.Decision = "deny"
+		audit.Reason = "autosync import failed"
+		s.recordAudit(audit)
+		return fmt.Errorf("autosync import: %w", err)
+	}
+	audit.Decision = "allow"
+	s.recordAudit(audit)
+	return nil
+}

internal/server/substore_sync_test.go

@@ -0,0 +1,144 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/LatticeNet/lattice-sdk/model"
+	"github.com/LatticeNet/lattice-server/internal/plugin"
+	"github.com/LatticeNet/lattice-server/internal/store"
+)
+
+func newSubStoreSyncTestServer(t *testing.T) *Server {
+	t.Helper()
+	st, err := store.Open("")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return newLinemetaTestServer(t, st)
+}
+
+func seedSubStoreSecrets(t *testing.T, srv *Server, endpoint, autosync string) {
+	t.Helper()
+	if endpoint != "" {
+		if err := srv.store.PutPluginSecret(model.KVEntry{Bucket: pluginSecretBucketPrefix + subStorePluginID, Key: "endpoint", Value: endpoint}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if autosync != "" {
+		if err := srv.store.PutPluginSecret(model.KVEntry{Bucket: pluginSecretBucketPrefix + subStorePluginID, Key: "autosync", Value: autosync}); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestResolveSecretOperatorTargets(t *testing.T) {
+	srv := newSubStoreSyncTestServer(t)
+	seedSubStoreSecrets(t, srv, "https://sub.example.com/api/token/abc", "")
+	fields := []string{"base_url"}
+
+	// secret:// ref resolves and rewrites the payload.
+	out, err := srv.resolveSecretOperatorTargets(lineUserTestPrincipal(), subStorePluginID,
+		json.RawMessage(`{"base_url":"secret://endpoint","sub_name":"x"}`), fields)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got map[string]string
+	if err := json.Unmarshal(out, &got); err != nil {
+		t.Fatal(err)
+	}
+	if got["base_url"] != "https://sub.example.com/api/token/abc" || got["sub_name"] != "x" {
+		t.Fatalf("rewritten payload: %v", got)
+	}
+
+	// Plain URLs pass through untouched.
+	in := json.RawMessage(`{"base_url":"https://direct.example.com"}`)
+	out, err = srv.resolveSecretOperatorTargets(lineUserTestPrincipal(), subStorePluginID, in, fields)
+	if err != nil || string(out) != string(in) {
+		t.Fatalf("passthrough: out=%s err=%v", out, err)
+	}
+
+	// Unknown key fails loud.
+	if _, err := srv.resolveSecretOperatorTargets(lineUserTestPrincipal(), subStorePluginID,
+		json.RawMessage(`{"base_url":"secret://nope"}`), fields); err == nil || !strings.Contains(err.Error(), "not saved") {
+		t.Fatalf("missing secret: %v", err)
+	}
+	// Malformed refs fail loud.
+	if _, err := srv.resolveSecretOperatorTargets(lineUserTestPrincipal(), subStorePluginID,
+		json.RawMessage(`{"base_url":"secret://"}`), fields); err == nil {
+		t.Fatal("empty ref: want error")
+	}
+	// Secrets from another plugin's namespace are invisible.
+	if _, err := srv.resolveSecretOperatorTargets(lineUserTestPrincipal(), "latticenet.other",
+		json.RawMessage(`{"base_url":"secret://endpoint"}`), fields); err == nil {
+		t.Fatal("cross-plugin namespace: want error")
+	}
+}
+
+func TestSubStoreAutoSyncTarget(t *testing.T) {
+	srv := newSubStoreSyncTestServer(t)
+	if _, ok := srv.subStoreAutoSyncTarget(); ok {
+		t.Fatal("no secrets: want disabled")
+	}
+	seedSubStoreSecrets(t, srv, "https://sub.example.com", "")
+	if _, ok := srv.subStoreAutoSyncTarget(); ok {
+		t.Fatal("endpoint without autosync flag: want disabled")
+	}
+	seedSubStoreSecrets(t, srv, "", "0")
+	if _, ok := srv.subStoreAutoSyncTarget(); ok {
+		t.Fatal("autosync=0: want disabled")
+	}
+	seedSubStoreSecrets(t, srv, "", "1")
+	endpoint, ok := srv.subStoreAutoSyncTarget()
+	if !ok || endpoint != "https://sub.example.com" {
+		t.Fatalf("enabled: %q %v", endpoint, ok)
+	}
+}
+
+func TestRunSubStoreAutoSync(t *testing.T) {
+	srv := newSubStoreSyncTestServer(t)
+	srv.pluginRuntime = plugin.NewRuntimeManagerWithOptions(plugin.RuntimeManagerOptions{})
+
+	var calls atomic.Int32
+	var gotEndpoint, gotPayload string
+	srv.subStoreSync.invoke = func(_ context.Context, pluginID, service, method string, payload json.RawMessage, targets []string) ([]byte, error) {
+		calls.Add(1)
+		gotEndpoint = targets[0]
+		gotPayload = string(payload)
+		return json.RawMessage(`{"ok":true}`), nil
+	}
+
+	// Plugin inactive: no invocation.
+	seedSubStoreSecrets(t, srv, "https://sub.example.com", "1")
+	if err := srv.runSubStoreAutoSync(); err != nil {
+		t.Fatal(err)
+	}
+	if calls.Load() != 0 {
+		t.Fatal("inactive plugin must not be invoked")
+	}
+
+	// Active plugin: exactly one invocation with the saved endpoint bound.
+	if err := srv.store.UpsertPluginInstallation(model.PluginInstallation{ID: subStorePluginID, Status: model.PluginStatusActive}); err != nil {
+		t.Fatal(err)
+	}
+	if err := srv.runSubStoreAutoSync(); err != nil {
+		t.Fatal(err)
+	}
+	if calls.Load() != 1 {
+		t.Fatalf("calls = %d, want 1", calls.Load())
+	}
+	if gotEndpoint != "https://sub.example.com" || !strings.Contains(gotPayload, `"base_url":"https://sub.example.com"`) {
+		t.Fatalf("endpoint binding: %q payload %s", gotEndpoint, gotPayload)
+	}
+
+	// Failed invocation returns an error (and audits a deny).
+	srv.subStoreSync.invoke = func(context.Context, string, string, string, json.RawMessage, []string) ([]byte, error) {
+		return nil, context.DeadlineExceeded
+	}
+	if err := srv.runSubStoreAutoSync(); err == nil {
+		t.Fatal("failed import: want error")
+	}
+}