From 94a5ab1b59acf4d63f9d81ff2cade25b83ed5331 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sat, 25 Apr 2026 21:13:53 +0100 Subject: [PATCH] Enforce data policy at AI model boundary --- .../v6/internal/subsystems/ai-runtime.md | 9 + .../v6/internal/subsystems/cloud-paid.md | 7 + .../internal/subsystems/security-privacy.md | 6 +- internal/ai/chat/agentic.go | 16 ++ internal/ai/chat/agentic_additional_test.go | 28 +++ internal/ai/chat/agentic_final.go | 6 + internal/ai/chat/service.go | 41 ++-- .../resource_policy_sanitizer.go | 193 ++++++++++++++++++ .../resource_policy_sanitizer_test.go | 122 +++++++++++ internal/ai/providers/quickstart.go | 43 +++- internal/ai/providers/quickstart_test.go | 10 + internal/ai/service.go | 61 ++++-- internal/ai/service_coverage_imp_test.go | 43 ++++ 13 files changed, 547 insertions(+), 38 deletions(-) create mode 100644 internal/ai/modelboundary/resource_policy_sanitizer.go create mode 100644 internal/ai/modelboundary/resource_policy_sanitizer_test.go diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 5bf400171..ab4b4c24a 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -85,6 +85,10 @@ runtime cost control, and shared AI transport surfaces. represented only as aggregate posture and omitted from detailed prompt sections, while sensitive alert text is scrubbed through the shared unified-resource redaction helper before it reaches a non-local model. + The final provider-bound chat, Patrol, investigation, tool-result, and + hosted-quickstart requests must also pass through that same resource-policy + sanitizer immediately before transport, so later agentic turns cannot + reintroduce local-only identifiers after the original context export. 6. Keep AI resource and incident context aligned with the canonical unified-resource timeline before falling back to patrol-local change detectors 7. Keep platform assistant read/control claims aligned with `docs/release-control/v6/internal/PLATFORM_SUPPORT_MODEL.md`. New @@ -189,6 +193,11 @@ mobile-local adapter. That server-owned boundary must also own the real upstream vendor model selection explicitly through license-server configuration rather than through a baked runtime fallback, so model churn does not leak into the Pulse runtime or API contract. +Every hosted quickstart provider call must carry the `resource-policy-v1` +data-policy marker declaring client-side enforcement of aggregate-only +local-only handling and exact resource-policy redaction. The public proxy must +reject prompt relay requests that omit that marker so the Pulse-hosted route +cannot silently become an ungoverned external-model bypass. That same Patrol quickstart boundary is now server-authoritative end to end. `internal/ai/quickstart.go` must bootstrap before the first Patrol-only quickstart use, resolve the strongest server-verified runtime authority in diff --git a/docs/release-control/v6/internal/subsystems/cloud-paid.md b/docs/release-control/v6/internal/subsystems/cloud-paid.md index cba9388fa..47443e67e 100644 --- a/docs/release-control/v6/internal/subsystems/cloud-paid.md +++ b/docs/release-control/v6/internal/subsystems/cloud-paid.md @@ -110,6 +110,7 @@ cloud-specific enforcement rules. 87. `internal/cloudcp/stripe/grace_enforcer.go`, `internal/cloudcp/stripe/helpers.go`, `internal/cloudcp/stripe/reconciler.go`, `internal/cloudcp/stripe/webhook.go` 88. `internal/hosted/hosted_metrics.go`, `internal/hosted/reaper.go` 89. `pulse-pro:ops/pulse-cloud/audit/` +90. `pulse-pro:license-server/quickstart_proxy.go` ## Shared Boundaries @@ -755,6 +756,12 @@ runtime may send an execution identifier for one higher-level Patrol run, and the license server must treat repeated proxy calls that share that execution identifier as one commercial quickstart run rather than charging once per agentic provider turn. +That same public proxy boundary must accept hosted quickstart prompts only +when the runtime attaches the `resource-policy-v1` data-policy marker proving +client-side aggregate-only local-only handling and exact resource-policy +redaction. Requests without that marker are invalid, because the Pulse-hosted +route is commercial activation support, not an ungoverned external prompt +relay. That quickstart allowance is therefore activation support, not the main commercial pitch: self-hosted pricing and docs may promise Patrol-only quickstart runs with no API key for activated or trial-backed installs, but diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md index 67dd8f1a7..d98d532f6 100644 --- a/docs/release-control/v6/internal/subsystems/security-privacy.md +++ b/docs/release-control/v6/internal/subsystems/security-privacy.md @@ -106,7 +106,7 @@ visibility, and privacy controls to operators. 6. Keep the shared storage-directory and secure storage-file hardening helper aligned with the crypto manager plus control-plane magic-link key and store handling whenever runtime data-root ownership assumptions change. 7. Keep auth-env ingestion and shared fingerprint-verifier TLS defaults aligned whenever runtime auth loading or pinned-certificate transport behavior changes. 8. Keep the Data Handling settings surface neutral and non-commercial: it may show resource policy posture, local-only counts, and redaction coverage, but it must not advertise trials, upgrades, paid plans, or monitoring limits. -9. Keep operator-facing Data Handling posture aligned with runtime AI/context enforcement: `local-only` resource details must not be sent to external model prompts, and sensitive free-form alert text must use the shared resource-policy redaction helper before leaving the local trust boundary. +9. Keep operator-facing Data Handling posture aligned with runtime AI/context enforcement: `local-only` resource details must not be sent to external model prompts, and sensitive free-form alert, tool-result, investigation, and hosted-quickstart text must use the shared resource-policy redaction helper before leaving the local trust boundary. All provider-bound AI requests to non-local models must use the shared resource-policy sanitizer immediately before transport so later agentic turns cannot bypass the advertised handling posture. ## Current State @@ -139,6 +139,10 @@ policy posture. It may expose the canonical sensitivity, handling-boundary, and redaction counts that Pulse already applies to resources, but it must stay informational and non-commercial so free/self-hosted operators are not shown paywall, trial, upgrade, or monitoring-limit prompts inside a privacy surface. +That posture is now enforced at the AI provider boundary too: non-local model +requests must be sanitized from the same resource-policy metadata that powers +the Data Handling surface, and hosted quickstart requests must carry an +explicit `resource-policy-v1` marker before the public proxy accepts them. That shared settings boundary now also has an explicit split of responsibilities: `frontend-modern/src/components/Settings/useSystemSettingsState.ts` remains the canonical owner for telemetry, local-upgrade-metrics, and auth/privacy runtime diff --git a/internal/ai/chat/agentic.go b/internal/ai/chat/agentic.go index af463d8f8..425cec123 100644 --- a/internal/ai/chat/agentic.go +++ b/internal/ai/chat/agentic.go @@ -167,6 +167,9 @@ type AgenticLoop struct { // Budget checker called after each turn to enforce token spending limits budgetChecker func() error + + // Request sanitizer applied immediately before model-bound transport. + requestSanitizer func(providers.ChatRequest) providers.ChatRequest } // NewAgenticLoop creates a new agentic loop @@ -195,6 +198,15 @@ func (a *AgenticLoop) UpdateTools() { a.tools = append(tools, userQuestionTool()) } +// SetRequestSanitizer installs a model-bound request sanitizer. It is called +// after compaction and before every provider turn so tool results from earlier +// turns cannot bypass resource policy redaction. +func (a *AgenticLoop) SetRequestSanitizer(fn func(providers.ChatRequest) providers.ChatRequest) { + a.mu.Lock() + defer a.mu.Unlock() + a.requestSanitizer = fn +} + // SetOrgID sets the org scope used when validating approval decisions. func (a *AgenticLoop) SetOrgID(orgID string) { a.mu.Lock() @@ -284,6 +296,7 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me } providerName := a.providerName modelName := a.modelName + requestSanitizer := a.requestSanitizer a.mu.Unlock() // Record telemetry for loop iteration @@ -478,6 +491,9 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me if turn == 0 { emitWorkflowState(callback, "investigate", "Inspecting infrastructure context and deciding the next step.", a.currentFSMState(), "") } + if requestSanitizer != nil { + req = requestSanitizer(req) + } const maxProviderAttempts = 2 err := error(nil) diff --git a/internal/ai/chat/agentic_additional_test.go b/internal/ai/chat/agentic_additional_test.go index 81c34760a..172c37fe6 100644 --- a/internal/ai/chat/agentic_additional_test.go +++ b/internal/ai/chat/agentic_additional_test.go @@ -187,6 +187,34 @@ func TestEnsureFinalTextResponse(t *testing.T) { } } +func TestEnsureFinalTextResponseAppliesRequestSanitizer(t *testing.T) { + provider := &stubStreamingProvider{} + loop := &AgenticLoop{provider: provider, baseSystemPrompt: "raw-host"} + loop.SetRequestSanitizer(func(req providers.ChatRequest) providers.ChatRequest { + req.System = strings.ReplaceAll(req.System, "raw-host", "[redacted]") + req.Messages = append([]providers.Message(nil), req.Messages...) + for i := range req.Messages { + req.Messages[i].Content = strings.ReplaceAll(req.Messages[i].Content, "raw-host", "[redacted]") + } + return req + }) + + loop.ensureFinalTextResponse( + context.Background(), + "session-sanitized", + []Message{{Role: "assistant", Content: ""}}, + []providers.Message{{Role: "user", Content: "check raw-host"}}, + func(event StreamEvent) {}, + ) + + if strings.Contains(provider.lastRequest.System, "raw-host") { + t.Fatalf("summary system prompt was not sanitized: %q", provider.lastRequest.System) + } + if strings.Contains(provider.lastRequest.Messages[0].Content, "raw-host") { + t.Fatalf("summary message was not sanitized: %q", provider.lastRequest.Messages[0].Content) + } +} + func TestBuildAutomaticFallbackSummary(t *testing.T) { summary := buildAutomaticFallbackSummary([]Message{ {Role: "user", ToolResult: &ToolResult{ToolUseID: "pulse_query_0", Content: "nodes ok", IsError: false}}, diff --git a/internal/ai/chat/agentic_final.go b/internal/ai/chat/agentic_final.go index 8e3a1b3b4..be8596446 100644 --- a/internal/ai/chat/agentic_final.go +++ b/internal/ai/chat/agentic_final.go @@ -59,6 +59,12 @@ func (a *AgenticLoop) ensureFinalTextResponse( ToolChoice: &providers.ToolChoice{Type: providers.ToolChoiceNone}, // No Tools field — completely omit tools to prevent hallucinated function calls } + a.mu.Lock() + requestSanitizer := a.requestSanitizer + a.mu.Unlock() + if requestSanitizer != nil { + summaryReq = requestSanitizer(summaryReq) + } var summaryBuilder strings.Builder diff --git a/internal/ai/chat/service.go b/internal/ai/chat/service.go index 0d184444a..b77543a9b 100644 --- a/internal/ai/chat/service.go +++ b/internal/ai/chat/service.go @@ -12,6 +12,7 @@ import ( "github.com/google/uuid" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" + "github.com/rcourtman/pulse-go-rewrite/internal/ai/modelboundary" "github.com/rcourtman/pulse-go-rewrite/internal/ai/providers" "github.com/rcourtman/pulse-go-rewrite/internal/ai/tools" "github.com/rcourtman/pulse-go-rewrite/internal/config" @@ -77,23 +78,24 @@ type Config struct { type Service struct { mu sync.RWMutex - cfg *config.AIConfig - dataDir string - stateProvider StateProvider - readState unifiedresources.ReadState - agentServer AgentServer - executor *tools.PulseToolExecutor - actionAuditStore unifiedresources.ResourceStore - sessions *SessionStore - agenticLoop *AgenticLoop - provider providers.StreamingProvider - providerFactory func(modelStr string) (providers.StreamingProvider, error) - patrolProviderFactory func(modelStr string) (providers.StreamingProvider, error) - started bool - autonomousMode bool - contextPrefetcher *ContextPrefetcher - budgetChecker func() error // Optional mid-run budget enforcement - orgID string + cfg *config.AIConfig + dataDir string + stateProvider StateProvider + readState unifiedresources.ReadState + agentServer AgentServer + executor *tools.PulseToolExecutor + unifiedResourceProvider tools.UnifiedResourceProvider + actionAuditStore unifiedresources.ResourceStore + sessions *SessionStore + agenticLoop *AgenticLoop + provider providers.StreamingProvider + providerFactory func(modelStr string) (providers.StreamingProvider, error) + patrolProviderFactory func(modelStr string) (providers.StreamingProvider, error) + started bool + autonomousMode bool + contextPrefetcher *ContextPrefetcher + budgetChecker func() error // Optional mid-run budget enforcement + orgID string activeMu sync.RWMutex activeExecutions map[string]map[*AgenticLoop]struct{} @@ -451,6 +453,7 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac autonomousMode := false s.mu.RLock() baseExecutor := s.executor + unifiedResourceProvider := s.unifiedResourceProvider autonomousMode = s.autonomousMode if s.cfg != nil { configuredModel = strings.TrimSpace(s.cfg.GetChatModel()) @@ -495,6 +498,7 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac loop.SetOrgID(s.orgID) } loop.SetAutonomousMode(autonomousMode) + loop.SetRequestSanitizer(modelboundary.RequestSanitizerForModel(selectedModel, unifiedResourceProvider)) s.registerActiveLoop(session.ID, loop) defer s.unregisterActiveLoop(session.ID, loop) @@ -763,6 +767,7 @@ func (s *Service) ExecutePatrolStream(ctx context.Context, req PatrolRequest, ca } sessions := s.sessions baseExecutor := s.executor + unifiedResourceProvider := s.unifiedResourceProvider cfg := s.cfg s.mu.RUnlock() executor := baseExecutor @@ -797,6 +802,7 @@ func (s *Service) ExecutePatrolStream(ctx context.Context, req PatrolRequest, ca tempLoop.SetOrgID(s.orgID) tempLoop.SetAutonomousMode(true) // Patrol runs without approval prompts tempLoop.SetExecutionID(req.ExecutionID) + tempLoop.SetRequestSanitizer(modelboundary.RequestSanitizerForModel(patrolModel, unifiedResourceProvider)) if req.MaxTurns > 0 { tempLoop.SetMaxTurns(req.MaxTurns) } @@ -1264,6 +1270,7 @@ func (s *Service) SetKnowledgeStoreProvider(provider KnowledgeStoreProvider) { func (s *Service) SetUnifiedResourceProvider(provider tools.UnifiedResourceProvider) { s.mu.Lock() defer s.mu.Unlock() + s.unifiedResourceProvider = provider if s.executor != nil { s.executor.SetUnifiedResourceProvider(provider) } diff --git a/internal/ai/modelboundary/resource_policy_sanitizer.go b/internal/ai/modelboundary/resource_policy_sanitizer.go new file mode 100644 index 000000000..9e7be33b6 --- /dev/null +++ b/internal/ai/modelboundary/resource_policy_sanitizer.go @@ -0,0 +1,193 @@ +package modelboundary + +import ( + "strings" + + "github.com/rcourtman/pulse-go-rewrite/internal/ai/providers" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +// UnifiedResourceProvider is the minimal policy source needed to sanitize +// provider-bound model requests. +type UnifiedResourceProvider interface { + GetByType(t unifiedresources.ResourceType) []unifiedresources.Resource +} + +type allUnifiedResourceProvider interface { + GetAll() []unifiedresources.Resource +} + +// RequestSanitizerForModel returns a sanitizer for non-local model traffic. +// It is intentionally applied at the final provider transport boundary so +// later tool-result turns cannot bypass the resource-policy posture exported +// to the operator-facing Data Handling surface. +func RequestSanitizerForModel(model string, provider UnifiedResourceProvider) func(providers.ChatRequest) providers.ChatRequest { + if !ModelUsesExternalProvider(model) || provider == nil { + return nil + } + resources := resourcePolicySanitizerResources(provider) + if len(resources) == 0 { + return nil + } + return func(req providers.ChatRequest) providers.ChatRequest { + return sanitizeProviderRequestForResources(req, resources) + } +} + +// ModelUsesExternalProvider reports whether a model string routes outside the +// local Ollama trust boundary. +func ModelUsesExternalProvider(model string) bool { + model = strings.TrimSpace(model) + if model == "" { + return false + } + provider, _ := config.ParseModelString(model) + return provider != config.AIProviderOllama +} + +func resourcePolicySanitizerResources(provider UnifiedResourceProvider) []unifiedresources.Resource { + if provider == nil { + return nil + } + if allProvider, ok := provider.(allUnifiedResourceProvider); ok { + return resourcesWithPolicy(unifiedresources.RefreshCanonicalMetadataSlice(allProvider.GetAll())) + } + + resourceTypes := []unifiedresources.ResourceType{ + unifiedresources.ResourceTypeAgent, + unifiedresources.ResourceTypeVM, + unifiedresources.ResourceTypeSystemContainer, + unifiedresources.ResourceTypeAppContainer, + unifiedresources.ResourceTypeDockerService, + unifiedresources.ResourceTypeK8sCluster, + unifiedresources.ResourceTypeK8sNode, + unifiedresources.ResourceTypePod, + unifiedresources.ResourceTypeK8sDeployment, + unifiedresources.ResourceTypeStorage, + unifiedresources.ResourceTypePBS, + unifiedresources.ResourceTypePMG, + unifiedresources.ResourceTypeCeph, + unifiedresources.ResourceTypePhysicalDisk, + } + + var resources []unifiedresources.Resource + seen := make(map[string]struct{}) + for _, resourceType := range resourceTypes { + for _, resource := range provider.GetByType(resourceType) { + key := string(resource.Type) + "\x00" + resource.ID + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + resources = append(resources, resource) + } + } + return resourcesWithPolicy(unifiedresources.RefreshCanonicalMetadataSlice(resources)) +} + +func resourcesWithPolicy(resources []unifiedresources.Resource) []unifiedresources.Resource { + filtered := make([]unifiedresources.Resource, 0, len(resources)) + for _, resource := range resources { + if resource.Policy == nil { + continue + } + filtered = append(filtered, resource) + } + return filtered +} + +func sanitizeProviderRequestForResources(req providers.ChatRequest, resources []unifiedresources.Resource) providers.ChatRequest { + if len(resources) == 0 { + return req + } + req.System = sanitizeResourcePolicyText(req.System, resources) + + if len(req.Messages) > 0 { + req.Messages = append([]providers.Message(nil), req.Messages...) + for i := range req.Messages { + req.Messages[i] = sanitizeProviderMessageForResources(req.Messages[i], resources) + } + } + if len(req.Tools) > 0 { + req.Tools = append([]providers.Tool(nil), req.Tools...) + for i := range req.Tools { + req.Tools[i] = sanitizeProviderToolForResources(req.Tools[i], resources) + } + } + return req +} + +func sanitizeProviderMessageForResources(msg providers.Message, resources []unifiedresources.Resource) providers.Message { + msg.Content = sanitizeResourcePolicyText(msg.Content, resources) + msg.ReasoningContent = sanitizeResourcePolicyText(msg.ReasoningContent, resources) + if msg.ToolResult != nil { + toolResult := *msg.ToolResult + toolResult.Content = sanitizeResourcePolicyText(toolResult.Content, resources) + msg.ToolResult = &toolResult + } + if len(msg.ToolCalls) > 0 { + msg.ToolCalls = append([]providers.ToolCall(nil), msg.ToolCalls...) + for i := range msg.ToolCalls { + msg.ToolCalls[i].Input = sanitizeResourcePolicyMap(msg.ToolCalls[i].Input, resources) + } + } + return msg +} + +func sanitizeProviderToolForResources(tool providers.Tool, resources []unifiedresources.Resource) providers.Tool { + tool.Description = sanitizeResourcePolicyText(tool.Description, resources) + tool.InputSchema = sanitizeResourcePolicyMap(tool.InputSchema, resources) + return tool +} + +func sanitizeResourcePolicyText(value string, resources []unifiedresources.Resource) string { + if strings.TrimSpace(value) == "" || len(resources) == 0 { + return value + } + redacted := value + for _, resource := range resources { + redacted = unifiedresources.ResourcePolicyRedactedText(redacted, resource) + } + return redacted +} + +func sanitizeResourcePolicyMap(values map[string]interface{}, resources []unifiedresources.Resource) map[string]interface{} { + if len(values) == 0 { + return values + } + sanitized := make(map[string]interface{}, len(values)) + for key, value := range values { + sanitized[key] = sanitizeResourcePolicyValue(value, resources) + } + return sanitized +} + +func sanitizeResourcePolicyValue(value interface{}, resources []unifiedresources.Resource) interface{} { + switch typed := value.(type) { + case string: + return sanitizeResourcePolicyText(typed, resources) + case []string: + out := make([]string, len(typed)) + for i := range typed { + out[i] = sanitizeResourcePolicyText(typed[i], resources) + } + return out + case []interface{}: + out := make([]interface{}, len(typed)) + for i := range typed { + out[i] = sanitizeResourcePolicyValue(typed[i], resources) + } + return out + case map[string]interface{}: + return sanitizeResourcePolicyMap(typed, resources) + case map[string]string: + out := make(map[string]string, len(typed)) + for key, nested := range typed { + out[key] = sanitizeResourcePolicyText(nested, resources) + } + return out + default: + return value + } +} diff --git a/internal/ai/modelboundary/resource_policy_sanitizer_test.go b/internal/ai/modelboundary/resource_policy_sanitizer_test.go new file mode 100644 index 000000000..9391b278e --- /dev/null +++ b/internal/ai/modelboundary/resource_policy_sanitizer_test.go @@ -0,0 +1,122 @@ +package modelboundary + +import ( + "strings" + "testing" + + "github.com/rcourtman/pulse-go-rewrite/internal/ai/providers" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +type policySanitizerProvider struct { + resources []unifiedresources.Resource +} + +func (p policySanitizerProvider) GetAll() []unifiedresources.Resource { + return append([]unifiedresources.Resource(nil), p.resources...) +} + +func (p policySanitizerProvider) GetByType(t unifiedresources.ResourceType) []unifiedresources.Resource { + var out []unifiedresources.Resource + for _, resource := range p.resources { + if resource.Type == t { + out = append(out, resource) + } + } + return out +} + +func TestRequestSanitizerForModelRedactsExternalModelRequest(t *testing.T) { + resource := unifiedresources.Resource{ + ID: "agent/pve-secret", + Type: unifiedresources.ResourceTypeAgent, + Name: "pve-secret", + Tags: []string{"restricted"}, + Identity: unifiedresources.ResourceIdentity{ + Hostnames: []string{"pve-secret.lan"}, + IPAddresses: []string{"10.0.0.5"}, + ClusterName: "prod-alias", + }, + } + sanitizer := RequestSanitizerForModel("openai:gpt-4o", policySanitizerProvider{resources: []unifiedresources.Resource{resource}}) + if sanitizer == nil { + t.Fatal("expected external model sanitizer") + } + + req := providers.ChatRequest{ + System: "Investigate pve-secret at 10.0.0.5", + Messages: []providers.Message{ + { + Role: "user", + Content: "pve-secret has alerts on pve-secret.lan", + ToolCalls: []providers.ToolCall{{ + ID: "tool-1", + Input: map[string]interface{}{ + "target": "pve-secret", + "nested": map[string]interface{}{"host": "pve-secret.lan"}, + "hosts": []string{"pve-secret.lan"}, + }, + }}, + }, + { + Role: "tool", + ToolResult: &providers.ToolResult{ToolUseID: "tool-1", Content: "agent/pve-secret reports prod-alias"}, + }, + }, + Tools: []providers.Tool{{ + Name: "pulse_read", + Description: "Read pve-secret", + InputSchema: map[string]interface{}{ + "properties": map[string]interface{}{ + "target": map[string]interface{}{ + "enum": []interface{}{"pve-secret", "other"}, + }, + }, + }, + }}, + } + + got := sanitizer(req) + combined := got.System + "\n" + got.Messages[0].Content + "\n" + got.Messages[1].ToolResult.Content + "\n" + got.Tools[0].Description + rawValues := []string{"pve-secret", "10.0.0.5", "pve-secret.lan", "agent/pve-secret", "prod-alias"} + for _, raw := range rawValues { + if strings.Contains(combined, raw) { + t.Fatalf("sanitized request still contains %q: %s", raw, combined) + } + } + if gotTarget, _ := got.Messages[0].ToolCalls[0].Input["target"].(string); strings.Contains(gotTarget, "pve-secret") { + t.Fatalf("tool call input target was not redacted: %q", gotTarget) + } + nested := got.Messages[0].ToolCalls[0].Input["nested"].(map[string]interface{}) + if gotHost, _ := nested["host"].(string); strings.Contains(gotHost, "pve-secret.lan") { + t.Fatalf("nested tool call input host was not redacted: %q", gotHost) + } + hosts := got.Messages[0].ToolCalls[0].Input["hosts"].([]string) + if strings.Contains(hosts[0], "pve-secret.lan") { + t.Fatalf("tool call input host slice was not redacted: %q", hosts[0]) + } + toolProperties := got.Tools[0].InputSchema["properties"].(map[string]interface{}) + toolTarget := toolProperties["target"].(map[string]interface{}) + toolEnum := toolTarget["enum"].([]interface{}) + if gotEnum, _ := toolEnum[0].(string); strings.Contains(gotEnum, "pve-secret") { + t.Fatalf("tool schema enum was not redacted: %q", gotEnum) + } + if req.Messages[0].Content != "pve-secret has alerts on pve-secret.lan" { + t.Fatalf("sanitizer mutated original request content: %q", req.Messages[0].Content) + } + if req.Tools[0].Description != "Read pve-secret" { + t.Fatalf("sanitizer mutated original tool description: %q", req.Tools[0].Description) + } +} + +func TestRequestSanitizerForModelSkipsLocalModel(t *testing.T) { + resource := unifiedresources.Resource{ + ID: "agent/pve-secret", + Type: unifiedresources.ResourceTypeAgent, + Name: "pve-secret", + Policy: &unifiedresources.ResourcePolicy{Routing: unifiedresources.ResourceRoutingPolicy{Scope: unifiedresources.ResourceRoutingScopeLocalOnly}}, + } + if sanitizer := RequestSanitizerForModel("ollama:llama3", policySanitizerProvider{resources: []unifiedresources.Resource{resource}}); sanitizer != nil { + t.Fatal("expected no sanitizer for local Ollama model") + } +} diff --git a/internal/ai/providers/quickstart.go b/internal/ai/providers/quickstart.go index f24bc283a..2e67b64b7 100644 --- a/internal/ai/providers/quickstart.go +++ b/internal/ai/providers/quickstart.go @@ -24,6 +24,8 @@ const ( quickstartRequestTimeout = 300 * time.Second // 5 minutes quickstartMaxRetries = 2 quickstartInitialBackoff = 2 * time.Second + + quickstartDataPolicyVersion = "resource-policy-v1" ) // QuickstartClient implements the Provider interface for the Pulse-hosted @@ -80,14 +82,22 @@ func NewQuickstartClientWithToken(quickstartToken string, onStateSync func(Quick // quickstartRequest is the payload sent to the hosted proxy. type quickstartRequest struct { - Messages []Message `json:"messages"` - System string `json:"system,omitempty"` - Tools []Tool `json:"tools,omitempty"` - ExecutionID string `json:"execution_id,omitempty"` + Messages []Message `json:"messages"` + System string `json:"system,omitempty"` + Tools []Tool `json:"tools,omitempty"` + ExecutionID string `json:"execution_id,omitempty"` + DataPolicy quickstartDataPolicy `json:"data_policy"` // LicenseID is retained only for the legacy caller-chosen identity path. LicenseID string `json:"license_id,omitempty"` } +type quickstartDataPolicy struct { + Version string `json:"version"` + ClientEnforced bool `json:"client_enforced"` + LocalOnlyHandling string `json:"local_only_handling"` + ResourceRedactionMode string `json:"resource_redaction_mode"` +} + // quickstartResponse is the response from the hosted proxy. type quickstartResponse struct { Content string `json:"content"` @@ -170,6 +180,7 @@ func (c *QuickstartClient) Chat(ctx context.Context, req ChatRequest) (*ChatResp System: req.System, Tools: req.Tools, ExecutionID: strings.TrimSpace(req.ExecutionID), + DataPolicy: defaultQuickstartDataPolicy(), } if strings.TrimSpace(c.quickstartToken) == "" { payload.LicenseID = c.licenseID @@ -268,11 +279,18 @@ func (c *QuickstartClient) Chat(ctx context.Context, req ChatRequest) (*ChatResp // TestConnection validates connectivity to the quickstart proxy. func (c *QuickstartClient) TestConnection(ctx context.Context) error { // Simple connectivity check — send a minimal request. - requestBody := `{"messages":[]}` - if strings.TrimSpace(c.quickstartToken) == "" { - requestBody = `{"messages":[],"license_id":"` + c.licenseID + `"}` + payload := quickstartRequest{ + Messages: []Message{}, + DataPolicy: defaultQuickstartDataPolicy(), } - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, quickstartProxyURL(), bytes.NewReader([]byte(requestBody))) + if strings.TrimSpace(c.quickstartToken) == "" { + payload.LicenseID = c.licenseID + } + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("quickstart: marshal test request: %w", err) + } + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, quickstartProxyURL(), bytes.NewReader(body)) if err != nil { return fmt.Errorf("quickstart: create test request: %w", err) } @@ -291,6 +309,15 @@ func (c *QuickstartClient) TestConnection(ctx context.Context) error { return nil } +func defaultQuickstartDataPolicy() quickstartDataPolicy { + return quickstartDataPolicy{ + Version: quickstartDataPolicyVersion, + ClientEnforced: true, + LocalOnlyHandling: "aggregate_only", + ResourceRedactionMode: "resource_policy_exact_values", + } +} + // Name returns the provider name. func (c *QuickstartClient) Name() string { return "quickstart" diff --git a/internal/ai/providers/quickstart_test.go b/internal/ai/providers/quickstart_test.go index ac11348cc..06be227b5 100644 --- a/internal/ai/providers/quickstart_test.go +++ b/internal/ai/providers/quickstart_test.go @@ -18,12 +18,14 @@ func TestQuickstartProxyURL_Default(t *testing.T) { func TestQuickstartClientChat_UsesOverrideProxyURL(t *testing.T) { var seenLicenseID string + var seenDataPolicy quickstartDataPolicy server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var req quickstartRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { t.Fatalf("decode request: %v", err) } seenLicenseID = req.LicenseID + seenDataPolicy = req.DataPolicy w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(quickstartResponse{ Content: "hello", @@ -47,6 +49,9 @@ func TestQuickstartClientChat_UsesOverrideProxyURL(t *testing.T) { if seenLicenseID != "lic_test" { t.Fatalf("license_id=%q want lic_test", seenLicenseID) } + if seenDataPolicy.Version != quickstartDataPolicyVersion || !seenDataPolicy.ClientEnforced { + t.Fatalf("data_policy=%#v want enforced %s", seenDataPolicy, quickstartDataPolicyVersion) + } if resp.Content != "hello" { t.Fatalf("content=%q want hello", resp.Content) } @@ -56,6 +61,7 @@ func TestQuickstartClientWithToken_UsesBearerAuthAndSyncsServerState(t *testing. var seenAuthorization string var seenLicenseID string var seenExecutionID string + var seenDataPolicy quickstartDataPolicy var synced QuickstartServerState var syncCalls int remaining := 16 @@ -69,6 +75,7 @@ func TestQuickstartClientWithToken_UsesBearerAuthAndSyncsServerState(t *testing. seenAuthorization = r.Header.Get("Authorization") seenLicenseID = req.LicenseID seenExecutionID = req.ExecutionID + seenDataPolicy = req.DataPolicy w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(quickstartResponse{ Content: "hello", @@ -105,6 +112,9 @@ func TestQuickstartClientWithToken_UsesBearerAuthAndSyncsServerState(t *testing. if seenExecutionID != "patrol-run-123" { t.Fatalf("execution_id=%q want patrol-run-123", seenExecutionID) } + if seenDataPolicy.Version != quickstartDataPolicyVersion || seenDataPolicy.LocalOnlyHandling != "aggregate_only" { + t.Fatalf("data_policy=%#v want local-only aggregate policy", seenDataPolicy) + } if resp.Content != "hello" { t.Fatalf("content=%q want hello", resp.Content) } diff --git a/internal/ai/service.go b/internal/ai/service.go index 671bee9d7..816f3104a 100644 --- a/internal/ai/service.go +++ b/internal/ai/service.go @@ -27,6 +27,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/ai/infradiscovery" "github.com/rcourtman/pulse-go-rewrite/internal/ai/knowledge" "github.com/rcourtman/pulse-go-rewrite/internal/ai/memory" + "github.com/rcourtman/pulse-go-rewrite/internal/ai/modelboundary" "github.com/rcourtman/pulse-go-rewrite/internal/ai/providers" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" @@ -1802,6 +1803,10 @@ func (s *Service) QuickAnalysis(ctx context.Context, req QuickAnalysisRequest) ( if cfg != nil && cfg.PatrolModel != "" { model = cfg.PatrolModel } + sanitizerModel := model + if sanitizerModel == "" && cfg != nil { + sanitizerModel = cfg.GetChatModel() + } messages := []providers.Message{ { @@ -1819,11 +1824,16 @@ func (s *Service) QuickAnalysis(ctx context.Context, req QuickAnalysisRequest) ( executionID = uuid.NewString() } - resp, err := provider.Chat(ctx, providers.ChatRequest{ + chatReq := providers.ChatRequest{ Messages: messages, Model: model, ExecutionID: executionID, - }) + } + if sanitizer := s.requestSanitizerForModel(sanitizerModel); sanitizer != nil { + chatReq = sanitizer(chatReq) + } + + resp, err := provider.Chat(ctx, chatReq) if err != nil { return "", fmt.Errorf("Pulse Assistant analysis failed: %w", err) } @@ -1835,6 +1845,13 @@ func (s *Service) QuickAnalysis(ctx context.Context, req QuickAnalysisRequest) ( return resp.Content, nil } +func (s *Service) requestSanitizerForModel(model string) func(providers.ChatRequest) providers.ChatRequest { + s.mu.RLock() + urp := s.unifiedResourceProvider + s.mu.RUnlock() + return modelboundary.RequestSanitizerForModel(model, urp) +} + // GetConfig returns a copy of the current AI config func (s *Service) GetConfig() *config.AIConfig { s.mu.RLock() @@ -2089,6 +2106,7 @@ func (s *Service) Execute(ctx context.Context, req ExecuteRequest) (*ExecuteResp // Determine the model to use for this request modelString := s.getModelForRequest(req) + requestSanitizer := s.requestSanitizerForModel(modelString) // Create a provider for this specific model (supports multi-provider switching) provider, err := providers.NewForModel(cfg, modelString) @@ -2170,13 +2188,18 @@ Always execute the commands rather than telling the user how to do it.` return nil, err } - resp, err := provider.Chat(ctx, providers.ChatRequest{ + chatReq := providers.ChatRequest{ Messages: messages, - Model: s.getModelForRequest(req), + Model: modelString, System: systemPrompt, MaxTokens: 4096, Tools: tools, - }) + } + if requestSanitizer != nil { + chatReq = requestSanitizer(chatReq) + } + + resp, err := provider.Chat(ctx, chatReq) if err != nil { return nil, fmt.Errorf("Pulse Assistant request failed: %w", err) } @@ -2289,6 +2312,7 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac // Determine the model to use for this request modelString := s.getModelForRequest(req) + requestSanitizer := s.requestSanitizerForModel(modelString) // Create a provider for this specific model (supports multi-provider switching) provider, err := providers.NewForModel(cfg, modelString) @@ -2401,13 +2425,18 @@ Always execute the commands rather than telling the user how to do it.` return nil, err } - resp, err := provider.Chat(ctx, providers.ChatRequest{ + chatReq := providers.ChatRequest{ Messages: messages, - Model: s.getModelForRequest(req), + Model: modelString, System: systemPrompt, MaxTokens: 4096, Tools: tools, - }) + } + if requestSanitizer != nil { + chatReq = requestSanitizer(chatReq) + } + + resp, err := provider.Chat(ctx, chatReq) if err != nil { log.Error().Err(err).Int("iteration", iteration).Msg("AI provider call failed") callback(StreamEvent{Type: "error", Data: map[string]string{"message": err.Error()}}) @@ -3568,11 +3597,15 @@ func (s *Service) AnalyzeForDiscovery(ctx context.Context, prompt string) (strin const discoveryResponseTokenBudget = 8192 // Make the API call - resp, err := provider.Chat(ctx, providers.ChatRequest{ + discoveryReq := providers.ChatRequest{ Messages: messages, Model: model, MaxTokens: discoveryResponseTokenBudget, // Discovery responses need room for structured JSON - }) + } + if sanitizer := s.requestSanitizerForModel(model); sanitizer != nil { + discoveryReq = sanitizer(discoveryReq) + } + resp, err := provider.Chat(ctx, discoveryReq) // If the primary provider fails (e.g., rate limited), try other configured providers if err != nil { @@ -3603,11 +3636,15 @@ func (s *Service) AnalyzeForDiscovery(ctx context.Context, prompt string) (strin Str("fallback_model", altModel). Msg("[Discovery] Primary provider failed, trying fallback") - resp, err = altProvider.Chat(ctx, providers.ChatRequest{ + altReq := providers.ChatRequest{ Messages: messages, Model: altModel, MaxTokens: discoveryResponseTokenBudget, - }) + } + if sanitizer := s.requestSanitizerForModel(altModel); sanitizer != nil { + altReq = sanitizer(altReq) + } + resp, err = altProvider.Chat(ctx, altReq) if err == nil { model = altModel provider = altProvider diff --git a/internal/ai/service_coverage_imp_test.go b/internal/ai/service_coverage_imp_test.go index a3869e7cb..2e6de25d4 100644 --- a/internal/ai/service_coverage_imp_test.go +++ b/internal/ai/service_coverage_imp_test.go @@ -65,6 +65,49 @@ func TestService_QuickAnalysis(t *testing.T) { } } +func TestService_QuickAnalysisSanitizesExternalModelRequest(t *testing.T) { + svc := NewService(nil, nil) + resource := unifiedresources.Resource{ + ID: "agent/pve-secret", + Type: unifiedresources.ResourceTypeAgent, + Name: "pve-secret", + Tags: []string{"restricted"}, + Identity: unifiedresources.ResourceIdentity{ + Hostnames: []string{"pve-secret.lan"}, + IPAddresses: []string{"10.0.0.5"}, + }, + } + svc.SetUnifiedResourceProvider(&mockUnifiedResourceProvider{ + getAllFunc: func() []unifiedresources.Resource { + return []unifiedresources.Resource{resource} + }, + }) + + var seen providers.ChatRequest + svc.provider = &mockProvider{ + chatFunc: func(ctx context.Context, req providers.ChatRequest) (*providers.ChatResponse, error) { + seen = req + return &providers.ChatResponse{Content: "ok"}, nil + }, + } + svc.cfg = &config.AIConfig{ + Enabled: true, + PatrolModel: "openai:gpt-4o", + } + + if _, err := svc.QuickAnalysis(context.Background(), QuickAnalysisRequest{ + Prompt: "check pve-secret at pve-secret.lan and 10.0.0.5", + }); err != nil { + t.Fatalf("QuickAnalysis failed: %v", err) + } + combined := seen.Messages[0].Content + "\n" + seen.Messages[1].Content + for _, raw := range []string{"pve-secret", "pve-secret.lan", "10.0.0.5"} { + if strings.Contains(combined, raw) { + t.Fatalf("provider request still contains %q: %s", raw, combined) + } + } +} + func TestService_AnalyzeForDiscovery(t *testing.T) { svc := NewService(nil, nil)