diff --git a/internal/server/lineusers.go b/internal/server/lineusers.go index 965e1b7..07f271a 100644 --- a/internal/server/lineusers.go +++ b/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 } diff --git a/internal/server/plugin_host.go b/internal/server/plugin_host.go index 3839d92..d7a83eb 100644 --- a/internal/server/plugin_host.go +++ b/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 diff --git a/internal/server/server.go b/internal/server/server.go index 52f521c..e6247b0 100644 --- a/internal/server/server.go +++ b/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) }, diff --git a/internal/server/server_plugin_invoke.go b/internal/server/server_plugin_invoke.go index 255b357..a3a65ca 100644 --- a/internal/server/server_plugin_invoke.go +++ b/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 { diff --git a/internal/server/substore_sync.go b/internal/server/substore_sync.go new file mode 100644 index 0000000..4dcec28 --- /dev/null +++ b/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://, resolved from the +// plugin's own encrypted secret bucket (pluginsecret:) 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 +} diff --git a/internal/server/substore_sync_test.go b/internal/server/substore_sync_test.go new file mode 100644 index 0000000..206b4d5 --- /dev/null +++ b/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") + } +} diff --git a/internal/server/vpnusers.go b/internal/server/vpnusers.go index e983a3b..394028e 100644 --- a/internal/server/vpnusers.go +++ b/internal/server/vpnusers.go @@ -244,6 +244,19 @@ func (s *Server) vpnCoreUsersRPC(_ context.Context, method string, request []byt // ── RPC: writes (proxy:admin) ───────────────────────────────────────────────── func (s *Server) vpnCoreUsersAdminRPC(ctx context.Context, method string, request []byte) ([]byte, error) { + out, err := s.vpnCoreUsersAdminDispatch(ctx, method, request) + // design-15 §7: every committed mutation re-arms the Sub-Store auto-sync; + // plan methods only queue approvals and do not change subscription content. + if err == nil { + switch method { + case "create", "update", "delete", "bind", "unbind", "rotate": + s.triggerVPNCoreMutation() + } + } + return out, err +} + +func (s *Server) vpnCoreUsersAdminDispatch(ctx context.Context, method string, request []byte) ([]byte, error) { switch method { case "create": return s.vpnUserCreate(request)