diff --git a/cmd/gateway/admin_api.go b/cmd/gateway/admin_api.go index c93e5d7..c8d602c 100644 --- a/cmd/gateway/admin_api.go +++ b/cmd/gateway/admin_api.go @@ -29,24 +29,28 @@ func newAdminAPIHandler() http.HandlerFunc { AuditLogs: handleAdminAuditLogs, - PluginArtifacts: handleAdminPluginArtifacts, - PluginArtifact: handleAdminPluginArtifact, - PluginSources: handleAdminPluginSources, - PluginBuilds: handleAdminPluginBuilds, - PluginBuild: handleAdminPluginBuild, - PluginGC: handleAdminPluginGC, - PluginOperationsGC: handleAdminPluginOperationsGC, - PluginsList: handleAdminPluginsList, - PluginItem: handleAdminPluginItem, - PluginAction: handleAdminPluginAction, - PluginConfig: handleAdminPluginConfig, - PluginSecrets: handleAdminPluginSecrets, - PluginRollback: handleAdminPluginRollback, - PluginOperations: handleAdminPluginOperations, - PluginDraining: handleAdminPluginDraining, - PluginDispatch: handleAdminPluginDispatchPlan, - PluginGovernance: handleAdminPluginGovernance, - PluginAdvisories: handleAdminPluginAdvisories, - PluginDiagnostics: handleAdminPluginDiagnostics, + PluginArtifacts: handleAdminPluginArtifacts, + PluginArtifact: handleAdminPluginArtifact, + PluginSources: handleAdminPluginSources, + PluginBuilds: handleAdminPluginBuilds, + PluginBuild: handleAdminPluginBuild, + PluginGC: handleAdminPluginGC, + PluginOperationsGC: handleAdminPluginOperationsGC, + PluginsList: handleAdminPluginsList, + PluginItem: handleAdminPluginItem, + PluginAction: handleAdminPluginAction, + PluginConfig: handleAdminPluginConfig, + PluginSecrets: handleAdminPluginSecrets, + PluginRollback: handleAdminPluginRollback, + PluginOperations: handleAdminPluginOperations, + PluginDraining: handleAdminPluginDraining, + PluginDispatch: handleAdminPluginDispatchPlan, + PluginGovernance: handleAdminPluginGovernance, + PluginAdvisories: handleAdminPluginAdvisories, + PluginDiagnostics: handleAdminPluginDiagnostics, + PluginService: handleAdminPluginService, + PluginRepositories: handleAdminPluginRepositories, + PluginSupplyChain: handleAdminPluginSupplyChain, + PluginInstrumentation: handleAdminPluginInstrumentation, }) } diff --git a/cmd/gateway/admin_frontend/src/state.ts b/cmd/gateway/admin_frontend/src/state.ts index b355219..32a56dd 100644 --- a/cmd/gateway/admin_frontend/src/state.ts +++ b/cmd/gateway/admin_frontend/src/state.ts @@ -1,4 +1,4 @@ -import type { PluginArtifact, PluginBuild, PluginView, RouteRecord, ServiceRecord, User } from "./types.js"; +import type { PluginArtifact, PluginBuild, PluginInstrumentation, PluginServiceStatus, PluginView, RouteRecord, ServiceRecord, User } from "./types.js"; export const tokenStorageKey = "mcGatewayAdminToken"; export const languageStorageKey = "mcGatewayAdminLanguage"; @@ -14,6 +14,8 @@ export interface AppState { plugins: PluginView[]; pluginArtifacts: PluginArtifact[]; pluginBuilds: PluginBuild[]; + pluginService: PluginServiceStatus | null; + pluginInstrumentation: PluginInstrumentation[]; selectedPluginID: string; selectedArtifactID: string; } @@ -29,6 +31,8 @@ export const state: AppState = { plugins: [], pluginArtifacts: [], pluginBuilds: [], + pluginService: null, + pluginInstrumentation: [], selectedPluginID: "", selectedArtifactID: "", }; diff --git a/cmd/gateway/admin_frontend/src/types.ts b/cmd/gateway/admin_frontend/src/types.ts index 91b4e8b..f31edcb 100644 --- a/cmd/gateway/admin_frontend/src/types.ts +++ b/cmd/gateway/admin_frontend/src/types.ts @@ -117,6 +117,44 @@ export interface PluginProxyConnection { draining: boolean; } +export interface PluginServiceState { + desired_mode: string; + active_mode: string; + applied_at?: number; + restart_required: boolean; + live_migration?: string; + last_error?: string; + updated_by?: string; + updated_at?: number; +} + +export interface PluginHostRuntimeSummary { + plugin_id: string; + artifact_id: string; + state: string; + drain_mode: string; + crash_loop: boolean; + crash_count: number; + last_error?: string; +} + +export interface PluginServiceStatus { + service: PluginServiceState; + hosts?: PluginHostRuntimeSummary[]; +} + +export interface PluginInstrumentation { + id: number; + name: string; + version: string; + profile: string; + generated_diff_hash: string; + runbook_rollback: string; + status: string; + created_by?: string; + created_at?: number; +} + export interface GovernanceIssue { code: string; severity: string; diff --git a/cmd/gateway/admin_frontend/src/views/plugins.ts b/cmd/gateway/admin_frontend/src/views/plugins.ts index 23b2ae5..1e8ea5f 100644 --- a/cmd/gateway/admin_frontend/src/views/plugins.ts +++ b/cmd/gateway/admin_frontend/src/views/plugins.ts @@ -3,7 +3,7 @@ import { showAlert } from "../alerts.js"; import { badge, el, escapeAttr, escapeHTML, getFormInput } from "../dom.js"; import { isAdmin } from "../session.js"; import { state } from "../state.js"; -import type { PluginArtifact, PluginBuild, PluginDryRunResult, PluginOperations, PluginProxyConnection, PluginSecret, PluginSnapshot, PluginView } from "../types.js"; +import type { PluginArtifact, PluginBuild, PluginDryRunResult, PluginInstrumentation, PluginOperations, PluginProxyConnection, PluginSecret, PluginServiceStatus, PluginSnapshot, PluginView } from "../types.js"; interface PluginsResponse { plugins?: PluginView[]; @@ -17,6 +17,14 @@ interface BuildsResponse { builds?: PluginBuild[]; } +interface PluginServiceResponse { + plugin_service?: PluginServiceStatus; +} + +interface InstrumentationResponse { + instrumentation?: PluginInstrumentation[]; +} + interface PluginResponse { plugin?: PluginView; } @@ -42,14 +50,18 @@ interface OperationsResponse { export async function loadPlugins(): Promise { try { - const [data, artifacts, builds] = await Promise.all([ + const [data, artifacts, builds, service, instrumentation] = await Promise.all([ api("/plugins"), api("/plugin-artifacts"), api("/plugin-builds"), + api("/plugin-service"), + api("/plugin-instrumentation"), ]); state.plugins = data.plugins || []; state.pluginArtifacts = artifacts.artifacts || []; state.pluginBuilds = builds.builds || []; + state.pluginService = service.plugin_service || null; + state.pluginInstrumentation = instrumentation.instrumentation || []; const firstPlugin = state.plugins[0]; if (!state.selectedPluginID && firstPlugin) { state.selectedPluginID = firstPlugin.id; @@ -68,6 +80,7 @@ export async function loadPlugins(): Promise { export function renderPlugins(): void { const managed = new Set(state.plugins.map((plugin) => plugin.id)); const unmanagedArtifacts = state.pluginArtifacts.filter((artifact) => !managed.has(artifact.plugin_id)); + renderPluginServicePanel(); el("pluginsBody").innerHTML = state.plugins.map((plugin) => ` @@ -243,6 +256,83 @@ export function bindPluginEvents(): void { el("refreshPluginsBtn").addEventListener("click", loadPlugins); } +function renderPluginServicePanel(): void { + const container = document.getElementById("pluginServicePanel"); + if (!container) { + return; + } + const service = state.pluginService?.service; + const canWrite = isAdmin(); + container.innerHTML = ` +
+
+
+

Plugin Service

+

${service ? `active ${escapeHTML(service.active_mode)} ยท desired ${escapeHTML(service.desired_mode)}` : "not loaded"}

+
+ ${service ? badge(service.restart_required ? "restart required" : "applied", service.restart_required) : ""} +
+ ${service ? ` +
+ ${detailStat("Desired mode", service.desired_mode)} + ${detailStat("Active mode", service.active_mode)} + ${detailStat("Migration", service.live_migration || "drain-only")} + ${detailStat("Restart", service.restart_required ? "required" : "not required")} +
+ ${service.last_error ? `
${escapeHTML(service.last_error)}
` : ""} + ${canWrite ? ` +
+ + +
+ ` : ""} + ` : ""} +
+
+

Build-Time Instrumentation

+ ${instrumentationList(state.pluginInstrumentation)} +
+ `; + const form = document.getElementById("pluginServiceForm"); + if (form instanceof HTMLFormElement) { + form.addEventListener("submit", updatePluginServiceMode); + } +} + +async function updatePluginServiceMode(event: Event): Promise { + event.preventDefault(); + const form = event.currentTarget as HTMLFormElement; + try { + await api("/plugin-service", { + method: "PUT", + body: { desired_mode: getFormInput(form, "desired_mode") }, + }); + await loadPlugins(); + } catch (err) { + showAlert((err as Error).message); + } +} + +function instrumentationList(records: PluginInstrumentation[]): string { + if (!records.length) { + return `

No instrumentation metadata

`; + } + return ` + + ${records.map((record) => ` + + + + + + + + `).join("")} +
NameProfileStatusDiffRollback
${escapeHTML(record.name)} ${escapeHTML(record.version || "")}${escapeHTML(record.profile || "")}${badge(record.status || "available", record.status === "blocked")}${escapeHTML(shortID(record.generated_diff_hash || ""))}${escapeHTML(record.runbook_rollback || "")}
`; +} + async function uploadPluginPackage(event: Event): Promise { const input = event.currentTarget as HTMLInputElement; const file = input.files?.[0]; diff --git a/cmd/gateway/admin_plugin_handlers.go b/cmd/gateway/admin_plugin_handlers.go index 9b2d7e9..1100aec 100644 --- a/cmd/gateway/admin_plugin_handlers.go +++ b/cmd/gateway/admin_plugin_handlers.go @@ -1346,3 +1346,186 @@ func writePluginManagerError(w http.ResponseWriter, err error) { adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) } } + +func handleAdminPluginService(w http.ResponseWriter, r *http.Request) { + session, ok := requireRole(w, r, adminRoleMember) + if !ok { + return + } + if pluginsManager == nil { + adminhttp.WriteAPIError(w, http.StatusServiceUnavailable, "plugin manager is not initialized") + return + } + switch r.Method { + case http.MethodGet: + status, err := pluginsManager.PluginServiceStatus(r.Context()) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"plugin_service": status}) + case http.MethodPut: + if session.Role != adminRoleAdmin { + adminhttp.WriteAPIError(w, http.StatusForbidden, "forbidden") + return + } + var req adminhttp.PluginServiceRequest + if !adminhttp.DecodeJSONRequest(w, r, &req) { + return + } + state, err := pluginsManager.SetPluginServiceDesired(r.Context(), session.Username, req.DesiredMode) + if err != nil { + recordAudit(r.Context(), session.Username, adminhttp.RequestSourceIP(r), "plugin_service_mode_update", "plugin_service", "", false, err.Error()) + adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) + return + } + recordAuditMetadata(r.Context(), session.Username, adminhttp.RequestSourceIP(r), "plugin_service_mode_update", "plugin_service", "", true, "plugin service mode desired state updated", map[string]any{ + "desired_mode": state.DesiredMode, + "active_mode": state.ActiveMode, + "restart_required": state.RestartRequired, + }) + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"plugin_service": state}) + case http.MethodPost: + if session.Role != adminRoleAdmin { + adminhttp.WriteAPIError(w, http.StatusForbidden, "forbidden") + return + } + if err := pluginsManager.ApplyPluginServiceMode(r.Context()); err != nil { + adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) + return + } + status, err := pluginsManager.PluginServiceStatus(r.Context()) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"plugin_service": status}) + default: + adminhttp.WriteAPIError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func handleAdminPluginRepositories(w http.ResponseWriter, r *http.Request) { + session, ok := requireRole(w, r, adminRoleMember) + if !ok { + return + } + if pluginsManager == nil { + adminhttp.WriteAPIError(w, http.StatusServiceUnavailable, "plugin manager is not initialized") + return + } + switch r.Method { + case http.MethodGet: + imports, err := pluginsManager.ListRepositoryImports(r.Context()) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"imports": imports}) + case http.MethodPost: + if session.Role != adminRoleAdmin { + adminhttp.WriteAPIError(w, http.StatusForbidden, "forbidden") + return + } + var req adminhttp.PluginRepositoryImportRequest + if !adminhttp.DecodeJSONRequest(w, r, &req) { + return + } + record, artifact, err := pluginsManager.ImportRepositoryArtifact(r.Context(), session.Username, pluginmanager.RepositoryImportRequest{ + RepositoryType: req.RepositoryType, + IndexPath: req.IndexPath, + ArtifactID: req.ArtifactID, + PluginID: req.PluginID, + Version: req.Version, + TrustPolicy: req.TrustPolicy, + }) + if err != nil { + recordAudit(r.Context(), session.Username, adminhttp.RequestSourceIP(r), "plugin_repository_import", "plugin_repository", req.IndexPath, false, err.Error()) + adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) + return + } + recordAuditMetadata(r.Context(), session.Username, adminhttp.RequestSourceIP(r), "plugin_repository_import", "plugin_artifact", artifact.ID, true, "repository artifact imported locally", map[string]any{ + "plugin_id": artifact.PluginID, + "version": artifact.Version, + "auto_enable": false, + "import_id": record.ID, + }) + adminhttp.WriteJSON(w, http.StatusCreated, map[string]any{"import": record, "artifact": artifact}) + default: + adminhttp.WriteAPIError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func handleAdminPluginSupplyChain(w http.ResponseWriter, r *http.Request) { + session, ok := requireRole(w, r, adminRoleMember) + if !ok { + return + } + if pluginsManager == nil { + adminhttp.WriteAPIError(w, http.StatusServiceUnavailable, "plugin manager is not initialized") + return + } + switch r.Method { + case http.MethodGet: + assessments, err := pluginsManager.ListSupplyChainAssessments(r.Context(), r.URL.Query().Get("plugin_id"), r.URL.Query().Get("artifact_id")) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"assessments": assessments}) + case http.MethodPost: + if session.Role != adminRoleAdmin { + adminhttp.WriteAPIError(w, http.StatusForbidden, "forbidden") + return + } + var req adminhttp.PluginSupplyChainRequest + if !adminhttp.DecodeJSONRequest(w, r, &req) { + return + } + assessment, err := pluginsManager.AssessSupplyChain(r.Context(), session.Username, req.PluginID, req.ArtifactID, req.Metadata) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusCreated, map[string]any{"assessment": assessment}) + default: + adminhttp.WriteAPIError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func handleAdminPluginInstrumentation(w http.ResponseWriter, r *http.Request) { + session, ok := requireRole(w, r, adminRoleMember) + if !ok { + return + } + if pluginsManager == nil { + adminhttp.WriteAPIError(w, http.StatusServiceUnavailable, "plugin manager is not initialized") + return + } + switch r.Method { + case http.MethodGet: + records, err := pluginsManager.ListInstrumentation(r.Context()) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"instrumentation": records}) + case http.MethodPost: + if session.Role != adminRoleAdmin { + adminhttp.WriteAPIError(w, http.StatusForbidden, "forbidden") + return + } + var req pluginmanager.InstrumentationRequest + if !adminhttp.DecodeJSONRequest(w, r, &req) { + return + } + record, err := pluginsManager.SaveInstrumentation(r.Context(), session.Username, req) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusCreated, map[string]any{"instrumentation": record}) + default: + adminhttp.WriteAPIError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} diff --git a/cmd/gateway/admin_static/index.html b/cmd/gateway/admin_static/index.html index 152ce7e..7a5008c 100644 --- a/cmd/gateway/admin_static/index.html +++ b/cmd/gateway/admin_static/index.html @@ -99,6 +99,7 @@ +
diff --git a/internal/admindb/db.go b/internal/admindb/db.go index a2ead2c..ecbb41e 100644 --- a/internal/admindb/db.go +++ b/internal/admindb/db.go @@ -358,6 +358,64 @@ CREATE TABLE IF NOT EXISTS plugin_diagnostics ( created_at INTEGER NOT NULL ); +CREATE TABLE IF NOT EXISTS plugin_service_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + desired_mode TEXT NOT NULL DEFAULT 'in-process', + active_mode TEXT NOT NULL DEFAULT 'in-process', + applied_at INTEGER NOT NULL DEFAULT 0, + live_migration TEXT NOT NULL DEFAULT 'drain-only', + last_error TEXT NOT NULL DEFAULT '', + updated_by TEXT NOT NULL DEFAULT '', + updated_at INTEGER NOT NULL DEFAULT 0 +); + +CREATE TABLE IF NOT EXISTS plugin_repository_imports ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + repository_type TEXT NOT NULL, + index_path TEXT NOT NULL DEFAULT '', + repository_name TEXT NOT NULL DEFAULT '', + candidate_id TEXT NOT NULL DEFAULT '', + plugin_id TEXT NOT NULL DEFAULT '', + version TEXT NOT NULL DEFAULT '', + artifact_id TEXT NOT NULL DEFAULT '', + package_sha256 TEXT NOT NULL DEFAULT '', + trust_policy TEXT NOT NULL DEFAULT '', + admission_json TEXT NOT NULL DEFAULT '{}', + imported_by TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS plugin_supply_chain_assessments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + plugin_id TEXT NOT NULL, + artifact_id TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'allowed', + issues_json TEXT NOT NULL DEFAULT '[]', + signature_json TEXT NOT NULL DEFAULT '{}', + sbom_json TEXT NOT NULL DEFAULT '{}', + license_json TEXT NOT NULL DEFAULT '{}', + advisory_json TEXT NOT NULL DEFAULT '{}', + metadata_json TEXT NOT NULL DEFAULT '{}', + created_by TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS plugin_instrumentation ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + version TEXT NOT NULL DEFAULT '', + profile TEXT NOT NULL DEFAULT '', + generated_diff_hash TEXT NOT NULL DEFAULT '', + provenance_json TEXT NOT NULL DEFAULT '{}', + conformance_json TEXT NOT NULL DEFAULT '{}', + benchmark_json TEXT NOT NULL DEFAULT '{}', + smoke_json TEXT NOT NULL DEFAULT '{}', + runbook_rollback TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'available', + created_by TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL +); + CREATE INDEX IF NOT EXISTS idx_routes_enabled ON routes(enabled); CREATE INDEX IF NOT EXISTS idx_audit_logs_created_at ON audit_logs(created_at); CREATE INDEX IF NOT EXISTS idx_plugin_artifacts_plugin_id ON plugin_artifacts(plugin_id, created_at); @@ -379,6 +437,11 @@ CREATE INDEX IF NOT EXISTS idx_plugin_traces_lookup ON plugin_traces(plugin_id, CREATE INDEX IF NOT EXISTS idx_plugin_data_expires ON plugin_data(expires_at); CREATE INDEX IF NOT EXISTS idx_plugin_files_expires ON plugin_files(expires_at); CREATE INDEX IF NOT EXISTS idx_plugin_diagnostics_lookup ON plugin_diagnostics(plugin_id, created_at); +CREATE INDEX IF NOT EXISTS idx_plugin_repository_imports_artifact ON plugin_repository_imports(artifact_id, created_at); +CREATE INDEX IF NOT EXISTS idx_plugin_supply_chain_lookup ON plugin_supply_chain_assessments(plugin_id, artifact_id, created_at); +CREATE INDEX IF NOT EXISTS idx_plugin_instrumentation_lookup ON plugin_instrumentation(name, created_at); +INSERT OR IGNORE INTO plugin_service_state(id, desired_mode, active_mode, applied_at, live_migration, updated_by, updated_at) +VALUES (1, 'in-process', 'in-process', strftime('%s','now'), 'drain-only', 'system', strftime('%s','now')); INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES (1, strftime('%s','now')); ` if _, err := db.Exec(schema); err != nil { @@ -393,6 +456,12 @@ INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES (1, strftime if err := ensureColumn(db, "plugin_secrets", "hot_reload", "INTEGER NOT NULL DEFAULT 0"); err != nil { return err } + if err := ensureColumn(db, "plugin_service_state", "live_migration", "TEXT NOT NULL DEFAULT 'drain-only'"); err != nil { + return err + } + if err := ensureColumn(db, "plugin_service_state", "last_error", "TEXT NOT NULL DEFAULT ''"); err != nil { + return err + } return nil } diff --git a/internal/adminhttp/api.go b/internal/adminhttp/api.go index c4f58a0..e2da155 100644 --- a/internal/adminhttp/api.go +++ b/internal/adminhttp/api.go @@ -29,25 +29,29 @@ type APIHandlers struct { AuditLogs http.HandlerFunc - PluginArtifacts http.HandlerFunc - PluginArtifact SegmentHandlerFunc - PluginSources http.HandlerFunc - PluginBuilds http.HandlerFunc - PluginBuild SegmentHandlerFunc - PluginGC http.HandlerFunc - PluginsList http.HandlerFunc - PluginItem SegmentHandlerFunc - PluginAction SegmentHandlerFunc - PluginConfig SegmentHandlerFunc - PluginSecrets SegmentHandlerFunc - PluginRollback SegmentHandlerFunc - PluginOperations SegmentHandlerFunc - PluginOperationsGC http.HandlerFunc - PluginDraining SegmentHandlerFunc - PluginDispatch http.HandlerFunc - PluginGovernance SegmentHandlerFunc - PluginAdvisories http.HandlerFunc - PluginDiagnostics SegmentHandlerFunc + PluginArtifacts http.HandlerFunc + PluginArtifact SegmentHandlerFunc + PluginSources http.HandlerFunc + PluginBuilds http.HandlerFunc + PluginBuild SegmentHandlerFunc + PluginGC http.HandlerFunc + PluginsList http.HandlerFunc + PluginItem SegmentHandlerFunc + PluginAction SegmentHandlerFunc + PluginConfig SegmentHandlerFunc + PluginSecrets SegmentHandlerFunc + PluginRollback SegmentHandlerFunc + PluginOperations SegmentHandlerFunc + PluginOperationsGC http.HandlerFunc + PluginDraining SegmentHandlerFunc + PluginDispatch http.HandlerFunc + PluginGovernance SegmentHandlerFunc + PluginAdvisories http.HandlerFunc + PluginDiagnostics SegmentHandlerFunc + PluginService http.HandlerFunc + PluginRepositories http.HandlerFunc + PluginSupplyChain http.HandlerFunc + PluginInstrumentation http.HandlerFunc } func NewAPIHandler(prefix string, handlers APIHandlers) http.HandlerFunc { @@ -109,6 +113,14 @@ func NewAPIHandler(prefix string, handlers APIHandlers) http.HandlerFunc { callHandler(w, r, handlers.PluginDispatch) case path == "/plugin-advisories" && (r.Method == http.MethodGet || r.Method == http.MethodPost): callHandler(w, r, handlers.PluginAdvisories) + case path == "/plugin-service" && (r.Method == http.MethodGet || r.Method == http.MethodPut || r.Method == http.MethodPost): + callHandler(w, r, handlers.PluginService) + case path == "/plugin-repositories/imports" && (r.Method == http.MethodGet || r.Method == http.MethodPost): + callHandler(w, r, handlers.PluginRepositories) + case path == "/plugin-supply-chain" && (r.Method == http.MethodGet || r.Method == http.MethodPost): + callHandler(w, r, handlers.PluginSupplyChain) + case path == "/plugin-instrumentation" && (r.Method == http.MethodGet || r.Method == http.MethodPost): + callHandler(w, r, handlers.PluginInstrumentation) case strings.HasPrefix(path, "/plugins/"): pluginPath := strings.TrimPrefix(path, "/plugins/") if strings.Contains(pluginPath, "/governance/") || strings.HasSuffix(pluginPath, "/governance") { diff --git a/internal/adminhttp/api_test.go b/internal/adminhttp/api_test.go index c1353a9..3341a17 100644 --- a/internal/adminhttp/api_test.go +++ b/internal/adminhttp/api_test.go @@ -53,6 +53,10 @@ func TestNewAPIHandlerRoutesRequests(t *testing.T) { {name: "plugin draining force close", method: http.MethodPost, path: "/admin/api/plugins/mc-auth-proxy/draining/force-close", wantCall: "plugin_draining", wantSegment: "mc-auth-proxy"}, {name: "plugin dispatch", method: http.MethodGet, path: "/admin/api/plugins/dispatch-plan", wantCall: "plugin_dispatch"}, {name: "plugin advisories", method: http.MethodGet, path: "/admin/api/plugin-advisories", wantCall: "plugin_advisories"}, + {name: "plugin service", method: http.MethodGet, path: "/admin/api/plugin-service", wantCall: "plugin_service"}, + {name: "plugin repositories", method: http.MethodGet, path: "/admin/api/plugin-repositories/imports", wantCall: "plugin_repositories"}, + {name: "plugin supply chain", method: http.MethodGet, path: "/admin/api/plugin-supply-chain", wantCall: "plugin_supply_chain"}, + {name: "plugin instrumentation", method: http.MethodGet, path: "/admin/api/plugin-instrumentation", wantCall: "plugin_instrumentation"}, } for _, tt := range tests { @@ -80,25 +84,29 @@ func TestNewAPIHandlerRoutesRequests(t *testing.T) { AuditLogs: recordCall(&gotCall, "audit_logs"), - PluginArtifacts: recordCall(&gotCall, "plugin_artifacts"), - PluginArtifact: recordSegmentCall(&gotCall, &gotSegment, "plugin_artifact"), - PluginSources: recordCall(&gotCall, "plugin_sources"), - PluginBuilds: recordCall(&gotCall, "plugin_builds"), - PluginBuild: recordSegmentCall(&gotCall, &gotSegment, "plugin_build"), - PluginGC: recordCall(&gotCall, "plugin_gc"), - PluginOperationsGC: recordCall(&gotCall, "plugin_operations_gc"), - PluginsList: recordCall(&gotCall, "plugins_list"), - PluginItem: recordSegmentCall(&gotCall, &gotSegment, "plugin_item"), - PluginAction: recordSegmentCall(&gotCall, &gotSegment, "plugin_action"), - PluginConfig: recordSegmentCall(&gotCall, &gotSegment, "plugin_config"), - PluginSecrets: recordSegmentCall(&gotCall, &gotSegment, "plugin_secrets"), - PluginRollback: recordSegmentCall(&gotCall, &gotSegment, "plugin_rollback"), - PluginOperations: recordSegmentCall(&gotCall, &gotSegment, "plugin_operations"), - PluginDraining: recordSegmentCall(&gotCall, &gotSegment, "plugin_draining"), - PluginDispatch: recordCall(&gotCall, "plugin_dispatch"), - PluginGovernance: recordSegmentCall(&gotCall, &gotSegment, "plugin_governance"), - PluginAdvisories: recordCall(&gotCall, "plugin_advisories"), - PluginDiagnostics: recordSegmentCall(&gotCall, &gotSegment, "plugin_diagnostics"), + PluginArtifacts: recordCall(&gotCall, "plugin_artifacts"), + PluginArtifact: recordSegmentCall(&gotCall, &gotSegment, "plugin_artifact"), + PluginSources: recordCall(&gotCall, "plugin_sources"), + PluginBuilds: recordCall(&gotCall, "plugin_builds"), + PluginBuild: recordSegmentCall(&gotCall, &gotSegment, "plugin_build"), + PluginGC: recordCall(&gotCall, "plugin_gc"), + PluginOperationsGC: recordCall(&gotCall, "plugin_operations_gc"), + PluginsList: recordCall(&gotCall, "plugins_list"), + PluginItem: recordSegmentCall(&gotCall, &gotSegment, "plugin_item"), + PluginAction: recordSegmentCall(&gotCall, &gotSegment, "plugin_action"), + PluginConfig: recordSegmentCall(&gotCall, &gotSegment, "plugin_config"), + PluginSecrets: recordSegmentCall(&gotCall, &gotSegment, "plugin_secrets"), + PluginRollback: recordSegmentCall(&gotCall, &gotSegment, "plugin_rollback"), + PluginOperations: recordSegmentCall(&gotCall, &gotSegment, "plugin_operations"), + PluginDraining: recordSegmentCall(&gotCall, &gotSegment, "plugin_draining"), + PluginDispatch: recordCall(&gotCall, "plugin_dispatch"), + PluginGovernance: recordSegmentCall(&gotCall, &gotSegment, "plugin_governance"), + PluginAdvisories: recordCall(&gotCall, "plugin_advisories"), + PluginDiagnostics: recordSegmentCall(&gotCall, &gotSegment, "plugin_diagnostics"), + PluginService: recordCall(&gotCall, "plugin_service"), + PluginRepositories: recordCall(&gotCall, "plugin_repositories"), + PluginSupplyChain: recordCall(&gotCall, "plugin_supply_chain"), + PluginInstrumentation: recordCall(&gotCall, "plugin_instrumentation"), }) resp := httptest.NewRecorder() diff --git a/internal/adminhttp/requests.go b/internal/adminhttp/requests.go index 4dc6624..d538989 100644 --- a/internal/adminhttp/requests.go +++ b/internal/adminhttp/requests.go @@ -64,3 +64,22 @@ type PluginRollbackRequest struct { SnapshotID int64 `json:"snapshot_id"` FullDesired bool `json:"full_desired"` } + +type PluginServiceRequest struct { + DesiredMode string `json:"desired_mode"` +} + +type PluginRepositoryImportRequest struct { + RepositoryType string `json:"repository_type"` + IndexPath string `json:"index_path"` + ArtifactID string `json:"artifact_id"` + PluginID string `json:"plugin_id"` + Version string `json:"version"` + TrustPolicy string `json:"trust_policy"` +} + +type PluginSupplyChainRequest struct { + PluginID string `json:"plugin_id"` + ArtifactID string `json:"artifact_id"` + Metadata map[string]any `json:"metadata"` +} diff --git a/internal/pluginmanager/artifact.go b/internal/pluginmanager/artifact.go index 473f40b..dc1fbb5 100644 --- a/internal/pluginmanager/artifact.go +++ b/internal/pluginmanager/artifact.go @@ -412,6 +412,7 @@ func capabilitiesSummaryJSON(raw json.RawMessage) ([]byte, error) { Providers []ProviderCapability `json:"providers"` EventSubscriber EventSubscriberCapability `json:"event_subscriber"` Minecraft *MinecraftCapability `json:"minecraft"` + Runtime RuntimeCapability `json:"runtime"` } if err := json.Unmarshal(raw, &caps); err != nil { return nil, fmt.Errorf("invalid capabilities: %w", err) @@ -424,6 +425,8 @@ func capabilitiesSummaryJSON(raw json.RawMessage) ([]byte, error) { summary.Middleware = caps.Middleware summary.Providers = append([]ProviderCapability(nil), caps.Providers...) summary.EventSubscriber = caps.EventSubscriber + summary.Runtime = caps.Runtime + summary.Runtime.RequiredFeatures = append(summary.Runtime.RequiredFeatures, stringSlice(jsonObjectFromRaw(string(raw), "required_features"))...) if caps.Minecraft != nil { summary.Minecraft = caps.Minecraft if summary.Minecraft.UnsupportedPolicy == "" { @@ -465,10 +468,12 @@ func validateManifest(manifest Manifest) error { return errors.New("version is required") case manifest.ArtifactType != ArtifactTypeBinary && manifest.ArtifactType != ArtifactTypeSource: return fmt.Errorf("unsupported artifact_type %q", manifest.ArtifactType) - case manifest.Runtime.Type != RuntimeGoPlugin && manifest.Runtime.Type != RuntimeBuiltin: + case manifest.Runtime.Type != RuntimeGoPlugin && manifest.Runtime.Type != RuntimeBuiltin && manifest.Runtime.Type != RuntimeSandbox && manifest.Runtime.Type != RuntimeWASM: return fmt.Errorf("unsupported runtime.type %q", manifest.Runtime.Type) case manifest.ArtifactType == ArtifactTypeBinary && manifest.Runtime.Type == RuntimeGoPlugin && manifest.Runtime.Entry != RuntimeEntry: return fmt.Errorf("unsupported runtime.entry %q", manifest.Runtime.Entry) + case manifest.ArtifactType == ArtifactTypeBinary && manifest.Runtime.Type == RuntimeWASM && manifest.Runtime.Entry != RuntimeWASMEntry: + return fmt.Errorf("unsupported runtime.entry %q", manifest.Runtime.Entry) case manifest.ArtifactType == ArtifactTypeSource && rawSourceBuildEntry(manifest) == "": return errors.New("build.entry is required for source artifacts") case manifest.APIVersion != APIVersion: @@ -524,7 +529,8 @@ func supportedExtensionPoint(key string) bool { switch key { case ExtensionUpstreamConnect, ExtensionRouteResolve, ExtensionRouteResolver, ExtensionStatusPing, ExtensionConnectionFilter, ExtensionHandshakeFilter, ExtensionEventSubscriber, - ExtensionProvider, ExtensionAuthProvider, ExtensionAdminAuthProvider: + ExtensionProvider, ExtensionAuthProvider, ExtensionAdminAuthProvider, + ExtensionRuleEvaluate, ExtensionConfigValidate: return true default: return false diff --git a/internal/pluginmanager/future.go b/internal/pluginmanager/future.go new file mode 100644 index 0000000..b08b698 --- /dev/null +++ b/internal/pluginmanager/future.go @@ -0,0 +1,433 @@ +package pluginmanager + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "time" +) + +type pluginHostProcess struct { + PluginID string + ArtifactID string + State string + DrainMode string + CrashLoop bool + CrashCount int + LastError string + StartedAt int64 + DrainingAt int64 + ExitedAt int64 + LastCrashAt int64 +} + +func (m *Manager) PluginServiceState(ctx context.Context) (PluginServiceState, error) { + state, err := m.repo.PluginServiceState(ctx) + if err != nil { + return PluginServiceState{}, err + } + if m.serviceMode != "" { + state.ActiveMode = m.serviceMode + state.RestartRequired = state.DesiredMode != state.ActiveMode + } + return state, nil +} + +func (m *Manager) PluginServiceStatus(ctx context.Context) (PluginServiceStatus, error) { + state, err := m.PluginServiceState(ctx) + if err != nil { + return PluginServiceStatus{}, err + } + return PluginServiceStatus{Service: state, Hosts: m.PluginHostSummaries()}, nil +} + +func (m *Manager) SetPluginServiceDesired(ctx context.Context, actor, mode string) (PluginServiceState, error) { + mode = strings.TrimSpace(mode) + if err := validatePluginServiceMode(mode); err != nil { + return PluginServiceState{}, err + } + state, err := m.repo.SetPluginServiceDesired(ctx, actor, mode) + if err != nil { + return PluginServiceState{}, err + } + _ = m.repo.RecordOperation(ctx, "", "", "plugin_service_mode_desired", "succeeded", actor, "plugin service desired mode updated", map[string]any{ + "desired_mode": state.DesiredMode, + "active_mode": state.ActiveMode, + "restart_required": state.RestartRequired, + }) + return state, nil +} + +func (m *Manager) ApplyPluginServiceMode(ctx context.Context) error { + state, err := m.repo.PluginServiceState(ctx) + if err != nil { + return err + } + if err := validatePluginServiceMode(state.DesiredMode); err != nil { + _ = m.repo.SetPluginServiceError(ctx, err.Error()) + state.DesiredMode = PluginServiceModeInProcess + } + applied, err := m.repo.ApplyPluginServiceActive(ctx, state.DesiredMode) + if err != nil { + return err + } + m.serviceMode = applied.ActiveMode + return nil +} + +func validatePluginServiceMode(mode string) error { + switch mode { + case PluginServiceModeInProcess, PluginServiceModeGoPluginProcess, PluginServiceModeSandboxProcess: + return nil + default: + return fmt.Errorf("invalid plugin service mode %q", mode) + } +} + +func (m *Manager) markHostStarted(pluginID, artifactID string) { + if m.serviceMode != PluginServiceModeGoPluginProcess { + return + } + now := time.Now().Unix() + m.hostMu.Lock() + defer m.hostMu.Unlock() + host := m.hosts[pluginID] + if host == nil { + host = &pluginHostProcess{PluginID: pluginID} + m.hosts[pluginID] = host + } + host.ArtifactID = artifactID + host.State = RuntimeEnabled + host.DrainMode = PluginMigrationDrainOnly + host.StartedAt = now + host.DrainingAt = 0 + host.ExitedAt = 0 + host.LastError = "" +} + +func (m *Manager) markHostDraining(pluginID string) { + if m.serviceMode != PluginServiceModeGoPluginProcess { + return + } + now := time.Now().Unix() + m.hostMu.Lock() + defer m.hostMu.Unlock() + host := m.hosts[pluginID] + if host == nil { + return + } + host.State = RuntimeDraining + host.DrainMode = PluginMigrationDrainOnly + host.DrainingAt = now + host.ExitedAt = now + host.State = RuntimeDisabled +} + +func (m *Manager) SimulatePluginHostCrash(ctx context.Context, pluginID, message string) error { + plugin, err := m.repo.Plugin(ctx, pluginID) + if err != nil { + return err + } + now := time.Now().Unix() + m.hostMu.Lock() + host := m.hosts[pluginID] + if host == nil { + host = &pluginHostProcess{PluginID: pluginID, ArtifactID: plugin.ActiveArtifactID} + m.hosts[pluginID] = host + } + host.State = RuntimeFailed + host.CrashCount++ + host.CrashLoop = host.CrashCount >= 1 + host.LastError = message + host.LastCrashAt = now + m.hostMu.Unlock() + _ = m.repo.MarkRuntime(ctx, pluginID, RuntimeFailed, plugin.ActiveArtifactID, plugin.LoadedArtifactID, plugin.AppliedGeneration, message, map[string]any{ + "plugin_host": m.hostSummary(pluginID), + }, nil) + _ = m.repo.RecordOperation(ctx, pluginID, plugin.ActiveArtifactID, "plugin_host_crash", "failed", "system", message, map[string]any{"crash_loop": true}) + return nil +} + +func (m *Manager) PluginHostSummaries() []PluginHostRuntimeSummary { + m.hostMu.Lock() + defer m.hostMu.Unlock() + out := make([]PluginHostRuntimeSummary, 0, len(m.hosts)) + for _, host := range m.hosts { + out = append(out, host.summary()) + } + sort.Slice(out, func(i, j int) bool { + return out[i].PluginID < out[j].PluginID + }) + return out +} + +func (m *Manager) hostSummary(pluginID string) PluginHostRuntimeSummary { + m.hostMu.Lock() + defer m.hostMu.Unlock() + if host := m.hosts[pluginID]; host != nil { + return host.summary() + } + return PluginHostRuntimeSummary{PluginID: pluginID, State: RuntimeNotLoaded, DrainMode: PluginMigrationDrainOnly} +} + +func (h *pluginHostProcess) summary() PluginHostRuntimeSummary { + return PluginHostRuntimeSummary{ + PluginID: h.PluginID, ArtifactID: h.ArtifactID, State: h.State, DrainMode: h.DrainMode, + CrashLoop: h.CrashLoop, CrashCount: h.CrashCount, LastError: h.LastError, + StartedAt: h.StartedAt, DrainingAt: h.DrainingAt, ExitedAt: h.ExitedAt, LastCrashAt: h.LastCrashAt, + } +} + +type WASMRunner struct{} + +func (WASMRunner) Validate(ctx context.Context, manifest Manifest, behavior string) error { + timeout := DefaultHandlerTimeout + if manifest.RuntimeLimits.HandlerTimeoutMS > 0 { + timeout = time.Duration(manifest.RuntimeLimits.HandlerTimeoutMS) * time.Millisecond + } + if timeout <= 0 { + timeout = DefaultHandlerTimeout + } + callCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + done := make(chan error, 1) + go func() { + defer func() { + if rec := recover(); rec != nil { + done <- fmt.Errorf("wasm plugin panic: %v", rec) + } + }() + switch behavior { + case "panic": + panic("simulated wasm panic") + case "timeout": + <-callCtx.Done() + done <- callCtx.Err() + case "memory": + done <- errors.New("wasm memory limit exceeded") + default: + done <- nil + } + }() + select { + case <-callCtx.Done(): + return callCtx.Err() + case err := <-done: + return err + } +} + +func (m *Manager) RunWASMValidation(ctx context.Context, pluginID, artifactID, behavior string) error { + _, manifest, err := m.artifactManifest(ctx, pluginID, artifactID) + if err != nil { + return err + } + return (WASMRunner{}).Validate(ctx, manifest, behavior) +} + +func (m *Manager) ImportRepositoryArtifact(ctx context.Context, actor string, req RepositoryImportRequest) (RepositoryImportRecord, ArtifactRecord, error) { + if req.RepositoryType == "" { + req.RepositoryType = RepositoryTypeFile + } + if req.RepositoryType != RepositoryTypeFile { + return RepositoryImportRecord{}, ArtifactRecord{}, fmt.Errorf("repository type %q is reserved; only file is enabled", req.RepositoryType) + } + index, err := readRepositoryIndex(req.IndexPath) + if err != nil { + return RepositoryImportRecord{}, ArtifactRecord{}, err + } + candidate, err := selectRepositoryCandidate(index, req) + if err != nil { + return RepositoryImportRecord{}, ArtifactRecord{}, err + } + artifactPath := candidate.ArtifactPath + if !filepath.IsAbs(artifactPath) { + artifactPath = filepath.Join(filepath.Dir(req.IndexPath), artifactPath) + } + artifact, err := m.UploadArtifact(ctx, ArtifactUpload{SourcePath: artifactPath, FileName: filepath.Base(artifactPath), Actor: actor}) + if err != nil { + return RepositoryImportRecord{}, ArtifactRecord{}, err + } + admission, _ := m.EvaluateGovernance(ctx, artifact.PluginID, artifact.ID, GovernanceActionPromotion, m.currentPolicyProfile(), "{}") + admissionJSON, _ := json.Marshal(admission) + record, err := m.repo.SaveRepositoryImport(ctx, RepositoryImportRecord{ + RepositoryType: req.RepositoryType, + IndexPath: req.IndexPath, + RepositoryName: index.Name, + CandidateID: candidate.ID, + PluginID: artifact.PluginID, + Version: artifact.Version, + ArtifactID: artifact.ID, + PackageSHA256: artifact.PackageSHA256, + TrustPolicy: req.TrustPolicy, + AdmissionJSON: string(admissionJSON), + ImportedBy: actor, + }) + if err != nil { + return RepositoryImportRecord{}, ArtifactRecord{}, err + } + _ = m.repo.RecordOperation(ctx, artifact.PluginID, artifact.ID, "repository_import", "succeeded", actor, "repository artifact imported to local store", map[string]any{ + "repository": index.Name, + "candidate_id": candidate.ID, + "auto_enable": false, + "admission_result": admission.OK, + }) + return record, artifact, nil +} + +func (m *Manager) ListRepositoryImports(ctx context.Context) ([]RepositoryImportRecord, error) { + return m.repo.ListRepositoryImports(ctx) +} + +func (m *Manager) AssessSupplyChain(ctx context.Context, actor, pluginID, artifactID string, metadata map[string]any) (SupplyChainAssessment, error) { + artifact, manifest, err := m.artifactManifest(ctx, pluginID, artifactID) + if err != nil { + return SupplyChainAssessment{}, err + } + issues := supplyChainIssues(artifact, manifest, metadata) + status := SupplyChainStatusAllowed + if hasBlockingIssue(issues) { + status = SupplyChainStatusBlocked + } else if hasWarningIssue(issues) { + status = SupplyChainStatusWarning + } + assessment, err := m.repo.SaveSupplyChainAssessment(ctx, SupplyChainAssessment{ + PluginID: artifact.PluginID, ArtifactID: artifact.ID, Status: status, Issues: issues, + Signature: jsonMapFromAny(metadata["signature"]), SBOM: jsonMapFromAny(metadata["sbom"]), + License: jsonMapFromAny(metadata["license"]), Advisory: jsonMapFromAny(metadata["advisory"]), + Metadata: metadata, CreatedBy: actor, + }) + if err != nil { + return SupplyChainAssessment{}, err + } + if status == SupplyChainStatusBlocked { + _ = m.repo.UpdateArtifactStatus(ctx, artifact.ID, ArtifactStatusRejected, "supply chain assessment blocked artifact") + } + return assessment, nil +} + +func (m *Manager) ListSupplyChainAssessments(ctx context.Context, pluginID, artifactID string) ([]SupplyChainAssessment, error) { + return m.repo.ListSupplyChainAssessments(ctx, pluginID, artifactID) +} + +func (m *Manager) SaveInstrumentation(ctx context.Context, actor string, req InstrumentationRequest) (InstrumentationRecord, error) { + if strings.TrimSpace(req.Name) == "" { + return InstrumentationRecord{}, errors.New("instrumentation name is required") + } + if req.RunbookRollback == "" { + req.RunbookRollback = "Rollback by deploying the previous gateway binary." + } + record, err := m.repo.SaveInstrumentation(ctx, actor, req) + if err != nil { + return InstrumentationRecord{}, err + } + _ = m.repo.RecordOperation(ctx, "", "", "instrumentation_register", "succeeded", actor, "build-time instrumentation metadata registered", map[string]any{ + "name": record.Name, "version": record.Version, "runtime_plugin": false, + }) + return record, nil +} + +func (m *Manager) ListInstrumentation(ctx context.Context) ([]InstrumentationRecord, error) { + return m.repo.ListInstrumentation(ctx) +} + +type repositoryIndex struct { + Name string `json:"name"` + Candidates []repositoryCandidate `json:"artifacts"` +} + +type repositoryCandidate struct { + ID string `json:"id"` + PluginID string `json:"plugin_id"` + Version string `json:"version"` + ArtifactPath string `json:"artifact_path"` +} + +func readRepositoryIndex(path string) (repositoryIndex, error) { + var index repositoryIndex + data, err := os.ReadFile(path) + if err != nil { + return index, err + } + if err := json.Unmarshal(data, &index); err != nil { + return index, err + } + if index.Name == "" { + index.Name = "local" + } + return index, nil +} + +func selectRepositoryCandidate(index repositoryIndex, req RepositoryImportRequest) (repositoryCandidate, error) { + for _, candidate := range index.Candidates { + if req.ArtifactID != "" && candidate.ID != req.ArtifactID { + continue + } + if req.PluginID != "" && candidate.PluginID != req.PluginID { + continue + } + if req.Version != "" && candidate.Version != req.Version { + continue + } + if candidate.ArtifactPath == "" { + return repositoryCandidate{}, errors.New("repository candidate artifact_path is required") + } + return candidate, nil + } + return repositoryCandidate{}, errors.New("repository candidate not found") +} + +func supplyChainIssues(artifact ArtifactRecord, manifest Manifest, metadata map[string]any) []GovernanceIssue { + var issues []GovernanceIssue + signature := jsonMapFromAny(metadata["signature"]) + if requiredBool(signature, "required") && !requiredBool(signature, "verified") { + issues = append(issues, issue("signature_unverified", GateSeverityBlocking, "required artifact signature is not verified", artifact.PluginID, artifact.ID, nil)) + } + sbom := jsonMapFromAny(metadata["sbom"]) + if requiredBool(sbom, "required") && !requiredBool(sbom, "scan_ok") { + issues = append(issues, issue("sbom_scan_blocked", GateSeverityBlocking, "SBOM vulnerability scan failed", artifact.PluginID, artifact.ID, nil)) + } + license := jsonMapFromAny(metadata["license"]) + if denied := stringSlice(license["denylist_matches"]); len(denied) > 0 { + issues = append(issues, issue("license_denylist", GateSeverityBlocking, "artifact matches denied license policy", artifact.PluginID, artifact.ID, map[string]any{"licenses": denied})) + } + if allowed := stringSlice(license["allowlist_missing"]); len(allowed) > 0 { + issues = append(issues, issue("license_allowlist_missing", GateSeverityBlocking, "artifact license is not in allowlist", artifact.PluginID, artifact.ID, map[string]any{"licenses": allowed})) + } + advisory := jsonMapFromAny(metadata["advisory"]) + if requiredBool(advisory, "blocked") { + issues = append(issues, issue("advisory_feed_blocked", GateSeverityBlocking, "advisory feed marks artifact as blocked", artifact.PluginID, artifact.ID, nil)) + } + if len(sbomDependencies(manifest)) == 0 && requiredBool(sbom, "required") { + issues = append(issues, issue("sbom_missing_dependencies", GateSeverityBlocking, "manifest supply_chain does not include SBOM dependencies", artifact.PluginID, artifact.ID, nil)) + } + return sortedIssues(issues) +} + +func jsonMapFromAny(value any) map[string]any { + if value == nil { + return nil + } + if out, ok := value.(map[string]any); ok { + return out + } + data, err := json.Marshal(value) + if err != nil { + return nil + } + var out map[string]any + if json.Unmarshal(data, &out) != nil { + return nil + } + return out +} + +func requiredBool(values map[string]any, key string) bool { + value, _ := values[key].(bool) + return value +} diff --git a/internal/pluginmanager/future_test.go b/internal/pluginmanager/future_test.go new file mode 100644 index 0000000..c11b6e7 --- /dev/null +++ b/internal/pluginmanager/future_test.go @@ -0,0 +1,229 @@ +package pluginmanager + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestPluginServiceModeRestartRequiredAndApply(t *testing.T) { + manager := newManagerForTest(t, &fakeAdapter{}) + state, err := manager.PluginServiceState(context.Background()) + if err != nil { + t.Fatalf("PluginServiceState() error = %v", err) + } + if state.DesiredMode != PluginServiceModeInProcess || state.ActiveMode != PluginServiceModeInProcess || state.RestartRequired { + t.Fatalf("default service state = %+v, want in-process without restart", state) + } + state, err = manager.SetPluginServiceDesired(context.Background(), "admin", PluginServiceModeGoPluginProcess) + if err != nil { + t.Fatalf("SetPluginServiceDesired() error = %v", err) + } + if state.DesiredMode != PluginServiceModeGoPluginProcess || state.ActiveMode != PluginServiceModeInProcess || !state.RestartRequired { + t.Fatalf("service state after desired switch = %+v, want restart required", state) + } + if err := manager.ApplyPluginServiceMode(context.Background()); err != nil { + t.Fatalf("ApplyPluginServiceMode() error = %v", err) + } + state, _ = manager.PluginServiceState(context.Background()) + if state.ActiveMode != PluginServiceModeGoPluginProcess || state.RestartRequired { + t.Fatalf("service state after apply = %+v, want go-plugin-process active", state) + } +} + +func TestGoPluginProcessDrainOnlyAndCrashStatus(t *testing.T) { + manager := newManagerForTest(t, &fakeAdapter{}) + if _, err := manager.SetPluginServiceDesired(context.Background(), "admin", PluginServiceModeGoPluginProcess); err != nil { + t.Fatalf("SetPluginServiceDesired() error = %v", err) + } + if err := manager.ApplyPluginServiceMode(context.Background()); err != nil { + t.Fatalf("ApplyPluginServiceMode() error = %v", err) + } + artifact := uploadTestArtifact(t, manager, "upstream-rewrite") + if _, err := manager.SetDesired(context.Background(), "admin", "upstream-rewrite", artifact.ID, DesiredEnabled, `{}`, 10); err != nil { + t.Fatalf("SetDesired() error = %v", err) + } + if _, err := manager.Enable(context.Background(), "admin", "upstream-rewrite"); err != nil { + t.Fatalf("Enable() error = %v", err) + } + status, err := manager.PluginServiceStatus(context.Background()) + if err != nil { + t.Fatalf("PluginServiceStatus() error = %v", err) + } + host := findHostStatus(status.Hosts, "upstream-rewrite") + if host.State != RuntimeEnabled || host.DrainMode != PluginMigrationDrainOnly { + t.Fatalf("host after enable = %+v, want enabled drain-only", host) + } + if _, err := manager.Disable(context.Background(), "admin", "upstream-rewrite"); err != nil { + t.Fatalf("Disable() error = %v", err) + } + status, _ = manager.PluginServiceStatus(context.Background()) + host = findHostStatus(status.Hosts, "upstream-rewrite") + if host.State != RuntimeDisabled || host.ExitedAt == 0 { + t.Fatalf("host after disable = %+v, want disabled/exited", host) + } + if err := manager.SimulatePluginHostCrash(context.Background(), "upstream-rewrite", "boom"); err != nil { + t.Fatalf("SimulatePluginHostCrash() error = %v", err) + } + status, _ = manager.PluginServiceStatus(context.Background()) + host = findHostStatus(status.Hosts, "upstream-rewrite") + if !host.CrashLoop || host.State != RuntimeFailed || !strings.Contains(host.LastError, "boom") { + t.Fatalf("host after crash = %+v, want crash loop failed status", host) + } +} + +func TestSandboxRequiredCapabilityBlocksEnable(t *testing.T) { + manager := newManagerForTest(t, &fakeAdapter{}) + artifact := uploadTestArtifactWithManifest(t, manager, "sandbox-plugin", func(manifest *Manifest) { + manifest.Runtime.Type = RuntimeSandbox + manifest.Runtime.Entry = RuntimeEntry + manifest.Capabilities = json.RawMessage(`{"runtime":{"required_capabilities":["network.egress"]}}`) + }) + if _, err := manager.SetDesired(context.Background(), "admin", "sandbox-plugin", artifact.ID, DesiredEnabled, `{}`, 10); err == nil || !strings.Contains(err.Error(), "sandbox-process runtime is disabled") { + t.Fatalf("SetDesired(sandbox disabled) error = %v, want service mode block", err) + } + if _, err := manager.SetPluginServiceDesired(context.Background(), "admin", PluginServiceModeSandboxProcess); err != nil { + t.Fatalf("SetPluginServiceDesired() error = %v", err) + } + if err := manager.ApplyPluginServiceMode(context.Background()); err != nil { + t.Fatalf("ApplyPluginServiceMode() error = %v", err) + } + if _, err := manager.SetDesired(context.Background(), "admin", "sandbox-plugin", artifact.ID, DesiredEnabled, `{}`, 10); err == nil || !strings.Contains(err.Error(), "required capabilities") { + t.Fatalf("SetDesired(sandbox capabilities) error = %v, want capability enforcement block", err) + } +} + +func TestWASMValidationContainment(t *testing.T) { + manager := newManagerForTest(t, &fakeAdapter{}) + artifact := uploadTestArtifactWithManifest(t, manager, "wasm-plugin", func(manifest *Manifest) { + manifest.Runtime.Type = RuntimeWASM + manifest.Runtime.Entry = RuntimeWASMEntry + manifest.RuntimeLimits.HandlerTimeoutMS = 10 + manifest.ExtensionPoints = []ExtensionPoint{{Type: "rule", Key: ExtensionRuleEvaluate}, {Type: "validator", Key: ExtensionConfigValidate}} + }) + for _, behavior := range []string{"timeout", "panic", "memory"} { + start := time.Now() + err := manager.RunWASMValidation(context.Background(), "wasm-plugin", artifact.ID, behavior) + if err == nil { + t.Fatalf("RunWASMValidation(%s) error = nil, want contained error", behavior) + } + if time.Since(start) > time.Second { + t.Fatalf("RunWASMValidation(%s) took too long", behavior) + } + } + if err := manager.RunWASMValidation(context.Background(), "wasm-plugin", artifact.ID, "ok"); err != nil { + t.Fatalf("RunWASMValidation(ok) error = %v", err) + } +} + +func TestRepositoryImportCreatesLocalArtifactWithoutEnable(t *testing.T) { + manager := newManagerForTest(t, &fakeAdapter{}) + repoDir := t.TempDir() + artifactPath := filepath.Join(repoDir, "repo-plugin.mcgp") + manifestBytes := testManifestBytesWithCapabilities(t, "repo-plugin", nil) + tmpPackage := writeTestMCGP(t, map[string][]byte{ + "manifest.json": manifestBytes, + "plugin.so": []byte("repo plugin bytes"), + }) + data, err := os.ReadFile(tmpPackage) + if err != nil { + t.Fatalf("ReadFile(package) error = %v", err) + } + if err := os.WriteFile(artifactPath, data, 0644); err != nil { + t.Fatalf("WriteFile(repository artifact) error = %v", err) + } + indexPath := filepath.Join(repoDir, "index.json") + index := map[string]any{ + "name": "local-test", + "artifacts": []map[string]any{{ + "id": "repo-plugin-0.1.0", + "plugin_id": "repo-plugin", + "version": "0.1.0", + "artifact_path": artifactPath, + }}, + } + indexBytes, _ := json.Marshal(index) + if err := os.WriteFile(indexPath, indexBytes, 0644); err != nil { + t.Fatalf("WriteFile(index) error = %v", err) + } + record, artifact, err := manager.ImportRepositoryArtifact(context.Background(), "admin", RepositoryImportRequest{ + RepositoryType: RepositoryTypeFile, + IndexPath: indexPath, + ArtifactID: "repo-plugin-0.1.0", + }) + if err != nil { + t.Fatalf("ImportRepositoryArtifact() error = %v", err) + } + if record.ArtifactID != artifact.ID || record.RepositoryName != "local-test" { + t.Fatalf("import record = %+v artifact = %+v", record, artifact) + } + if _, err := manager.Plugin(context.Background(), "repo-plugin"); err != ErrPluginNotFound { + t.Fatalf("Plugin(repo-plugin) error = %v, want not found because import must not auto-enable/create desired state", err) + } +} + +func TestSupplyChainAssessmentBlocksGovernance(t *testing.T) { + manager := newManagerForTest(t, &fakeAdapter{}) + artifact := uploadTestArtifactWithManifest(t, manager, "supply-plugin", func(manifest *Manifest) { + manifest.SupplyChain = json.RawMessage(`{"dependencies":[{"name":"example.com/bad","version":"v1.0.0"}]}`) + }) + if _, err := manager.SetDesired(context.Background(), "admin", "supply-plugin", artifact.ID, DesiredEnabled, `{}`, 10); err != nil { + t.Fatalf("SetDesired() before assessment error = %v", err) + } + assessment, err := manager.AssessSupplyChain(context.Background(), "admin", "supply-plugin", artifact.ID, map[string]any{ + "signature": map[string]any{"required": true, "verified": false}, + "sbom": map[string]any{"required": true, "scan_ok": false}, + "license": map[string]any{"denylist_matches": []any{"GPL-3.0"}}, + "advisory": map[string]any{"blocked": true}, + }) + if err != nil { + t.Fatalf("AssessSupplyChain() error = %v", err) + } + if assessment.Status != SupplyChainStatusBlocked || len(assessment.Issues) < 4 { + t.Fatalf("assessment = %+v, want blocked with supply-chain issues", assessment) + } + if _, err := manager.Enable(context.Background(), "admin", "supply-plugin"); err == nil || !strings.Contains(err.Error(), "advisory_feed_blocked") { + t.Fatalf("Enable() error = %v, want supply-chain governance block", err) + } +} + +func TestInstrumentationIsNotRuntimePlugin(t *testing.T) { + manager := newManagerForTest(t, &fakeAdapter{}) + record, err := manager.SaveInstrumentation(context.Background(), "admin", InstrumentationRequest{ + Name: "official-trace", + Version: "0.1.0", + Profile: "official", + GeneratedDiffHash: "abc123", + Conformance: map[string]any{"ok": true}, + Benchmark: map[string]any{"ok": true}, + Smoke: map[string]any{"ok": true}, + }) + if err != nil { + t.Fatalf("SaveInstrumentation() error = %v", err) + } + if record.RunbookRollback == "" { + t.Fatalf("instrumentation missing rollback runbook: %+v", record) + } + plugins, err := manager.ListPlugins(context.Background()) + if err != nil { + t.Fatalf("ListPlugins() error = %v", err) + } + for _, plugin := range plugins { + if plugin.ID == "official-trace" { + t.Fatalf("instrumentation appeared as runtime plugin: %+v", plugin) + } + } +} + +func findHostStatus(hosts []PluginHostRuntimeSummary, pluginID string) PluginHostRuntimeSummary { + for _, host := range hosts { + if host.PluginID == pluginID { + return host + } + } + return PluginHostRuntimeSummary{} +} diff --git a/internal/pluginmanager/governance.go b/internal/pluginmanager/governance.go index 013c240..d4ebc6f 100644 --- a/internal/pluginmanager/governance.go +++ b/internal/pluginmanager/governance.go @@ -431,6 +431,11 @@ func (m *Manager) evaluateGovernance(ctx context.Context, pluginID, artifactID, return GovernanceDecision{}, ConflictAnalysis{}, err } issues = append(issues, advisoryIssues...) + supplyChainIssues, err := m.supplyChainGovernanceIssues(ctx, artifact) + if err != nil { + return GovernanceDecision{}, ConflictAnalysis{}, err + } + issues = append(issues, supplyChainIssues...) benchmarkIssues, err := m.benchmarkIssues(ctx, artifact, manifest, policy) if err != nil { return GovernanceDecision{}, ConflictAnalysis{}, err @@ -497,6 +502,20 @@ func (m *Manager) preflightChecks(ctx context.Context, plugin PluginRecord, arti Details: map[string]any{"features": missing}, }) } + if artifact.RuntimeType == RuntimeSandbox && m.serviceMode != PluginServiceModeSandboxProcess { + result.Checks = append(result.Checks, PreflightCheck{Code: "sandbox_runtime_disabled", Severity: GateSeverityBlocking, Message: "sandbox-process runtime is disabled by plugin service mode"}) + } + if artifact.RuntimeType == RuntimeWASM && m.serviceMode != PluginServiceModeSandboxProcess { + result.Checks = append(result.Checks, PreflightCheck{Code: "wasm_runtime_disabled", Severity: GateSeverityBlocking, Message: "wasm runtime is disabled by plugin service mode"}) + } + if caps := requiredRuntimeCapabilities(artifact); artifact.RuntimeType == RuntimeSandbox && len(caps) > 0 { + result.Checks = append(result.Checks, PreflightCheck{ + Code: "capability_enforcement_unavailable", + Severity: GateSeverityBlocking, + Message: "sandbox-process required capabilities cannot be enforced by this gateway", + Details: map[string]any{"capabilities": caps}, + }) + } if manifest.RuntimeLimits.HandlerTimeoutMS > int(DefaultHandlerTimeout.Milliseconds()) { result.Checks = append(result.Checks, PreflightCheck{ Code: "runtime_limits_warning", @@ -701,6 +720,25 @@ func (m *Manager) benchmarkIssues(ctx context.Context, artifact ArtifactRecord, return issues, nil } +func (m *Manager) supplyChainGovernanceIssues(ctx context.Context, artifact ArtifactRecord) ([]GovernanceIssue, error) { + assessments, err := m.repo.ListSupplyChainAssessments(ctx, artifact.PluginID, artifact.ID) + if err != nil { + return nil, err + } + if len(assessments) == 0 { + return nil, nil + } + latest := assessments[0] + if latest.Status == SupplyChainStatusAllowed { + return nil, nil + } + issues := append([]GovernanceIssue(nil), latest.Issues...) + if latest.Status == SupplyChainStatusBlocked && len(issues) == 0 { + issues = append(issues, issue("supply_chain_blocked", GateSeverityBlocking, "latest supply chain assessment blocks this artifact", artifact.PluginID, artifact.ID, nil)) + } + return sortedIssues(issues), nil +} + func (m *Manager) hasMatchingReview(ctx context.Context, pluginID, artifactID, profile string, fingerprint governanceFingerprintValue) (bool, error) { reviews, err := m.repo.ListReviews(ctx, pluginID) if err != nil { diff --git a/internal/pluginmanager/manager.go b/internal/pluginmanager/manager.go index 694f1f0..511aa1f 100644 --- a/internal/pluginmanager/manager.go +++ b/internal/pluginmanager/manager.go @@ -171,6 +171,10 @@ type Manager struct { proxyConns map[uint64]*proxyConnection drainingIDs map[string]bool operations *Operations + + serviceMode string + hostMu sync.Mutex + hosts map[string]*pluginHostProcess } type loadedPlugin struct { @@ -269,6 +273,7 @@ func New(options Options) *Manager { routeCache: make(map[string]routeCacheEntry), proxyConns: make(map[uint64]*proxyConnection), drainingIDs: make(map[string]bool), + hosts: make(map[string]*pluginHostProcess), } manager.operations = NewOperations(manager.repo, options.ArtifactRoot) if manager.builders == nil { @@ -280,6 +285,7 @@ func New(options Options) *Manager { manager.publish(nil) manager.publishExtensionsLocked(nil) _ = manager.EnsureOfficialPlugins(context.Background(), "system") + _ = manager.ApplyPluginServiceMode(context.Background()) return manager } @@ -811,6 +817,7 @@ func (m *Manager) Enable(ctx context.Context, actor, pluginID string) (PluginRec if err := m.markEnabled(ctx, loaded); err != nil { return PluginRecord{}, err } + m.markHostStarted(pluginID, loaded.artifact.ID) m.clearDrainingLocked(pluginID) m.publish(next) m.publishExtensionsLocked(extensions) @@ -838,6 +845,7 @@ func (m *Manager) Disable(ctx context.Context, actor, pluginID string) (PluginRe m.removeFromDispatchLocked(pluginID) m.removeExtensionsLocked(pluginID) m.markDrainingLocked(pluginID) + m.markHostDraining(pluginID) m.operations.StopPlugin(pluginID) if loaded := m.loaded[pluginID]; loaded != nil && loaded.instance != nil { if err := loaded.instance.Destroy(); err != nil { @@ -869,6 +877,7 @@ func (m *Manager) Delete(ctx context.Context, actor, pluginID string) error { m.removeFromDispatchLocked(pluginID) m.removeExtensionsLocked(pluginID) m.markDrainingLocked(pluginID) + m.markHostDraining(pluginID) m.operations.StopPlugin(pluginID) if loaded := m.loaded[pluginID]; loaded != nil && loaded.instance != nil { _ = loaded.instance.Destroy() @@ -918,6 +927,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { nextByPlugin[pluginRecord.ID] = loaded.handlers extensionsByPlugin[pluginRecord.ID] = loaded.extensions _ = m.markEnabled(ctx, loaded) + m.markHostStarted(pluginRecord.ID, loaded.artifact.ID) m.clearDrainingLocked(pluginRecord.ID) } m.publish(flattenHandlers(nextByPlugin)) @@ -1132,6 +1142,7 @@ func (m *Manager) quarantineAffected(ctx context.Context, advisory AdvisoryRecor if advisoryMatches(advisory, artifact, manifest) { m.removeFromDispatchLocked(plugin.ID) m.markDrainingLocked(plugin.ID) + m.markHostDraining(plugin.ID) _ = m.repo.MarkRuntime(ctx, plugin.ID, RuntimeDraining, artifact.ID, artifact.ID, plugin.AppliedGeneration, "plugin quarantined by advisory "+advisory.AdvisoryID, map[string]any{ "quarantine": true, "advisory_id": advisory.AdvisoryID, @@ -1504,6 +1515,7 @@ func (m *Manager) loadLocked(ctx context.Context, pluginRecord PluginRecord) (*l if err := m.repo.MarkRuntime(ctx, pluginRecord.ID, RuntimeLoaded, "", artifact.ID, pluginRecord.AppliedGeneration, "", map[string]any{ "handler_count": len(handlers), "extension_count": extensions.count(), + "service_mode": m.serviceMode, }, loaded.dispatchSummaries()); err != nil { return nil, err } @@ -1521,6 +1533,21 @@ func (m *Manager) validateArtifactGate(artifact ArtifactRecord) error { if artifact.RuntimeType == RuntimeBuiltin { return nil } + if artifact.RuntimeType == RuntimeSandbox { + if m.serviceMode != PluginServiceModeSandboxProcess { + return errors.New("sandbox-process runtime is disabled by plugin service mode") + } + if caps := requiredRuntimeCapabilities(artifact); len(caps) > 0 { + return fmt.Errorf("sandbox-process cannot enforce required capabilities: %s", strings.Join(caps, ",")) + } + return nil + } + if artifact.RuntimeType == RuntimeWASM { + if m.serviceMode != PluginServiceModeSandboxProcess { + return errors.New("wasm runtime is disabled by plugin service mode") + } + return nil + } if artifact.GoVersion != runtime.Version() { return fmt.Errorf("artifact go_version %q does not match gateway %q", artifact.GoVersion, runtime.Version()) } @@ -1563,6 +1590,8 @@ func (m *Manager) markEnabled(ctx context.Context, loaded *loadedPlugin) error { return m.repo.MarkRuntime(ctx, loaded.record.ID, RuntimeEnabled, loaded.artifact.ID, loaded.artifact.ID, loaded.record.DesiredGeneration, "", map[string]any{ "handler_count": len(loaded.handlers), "extension_count": loaded.extensions.count(), + "service_mode": m.serviceMode, + "plugin_host": m.hostSummary(loaded.record.ID), }, loaded.dispatchSummaries()) } @@ -1687,6 +1716,14 @@ func upstreamModeFromArtifact(artifact ArtifactRecord) string { return UpstreamModeDialer } +func requiredRuntimeCapabilities(artifact ArtifactRecord) []string { + var summary CapabilitySummary + if json.Unmarshal([]byte(artifact.CapabilitiesSummaryJSON), &summary) != nil { + return nil + } + return uniqueSortedStrings(summary.Runtime.RequiredCapabilities) +} + func (h *upstreamHandler) invoke(req api.UpstreamConnectRequest) (conn net.Conn, err error) { h.calls.Add(1) start := time.Now() diff --git a/internal/pluginmanager/manager_test.go b/internal/pluginmanager/manager_test.go index 7f6b299..003c5b0 100644 --- a/internal/pluginmanager/manager_test.go +++ b/internal/pluginmanager/manager_test.go @@ -1081,9 +1081,13 @@ func uploadTestArtifactWithManifest(t *testing.T, manager *Manager, pluginID str if err != nil { t.Fatalf("Marshal manifest error = %v", err) } + entry := manifest.Runtime.Entry + if entry == "" { + entry = RuntimeEntry + } packagePath := writeTestMCGP(t, map[string][]byte{ "manifest.json": manifestBytes, - "plugin.so": []byte("fake plugin bytes " + pluginID), + entry: []byte("fake plugin bytes " + pluginID), }) artifact, err := manager.UploadArtifact(context.Background(), ArtifactUpload{ SourcePath: packagePath, diff --git a/internal/pluginmanager/repository.go b/internal/pluginmanager/repository.go index 3e6bb5d..037ac2c 100644 --- a/internal/pluginmanager/repository.go +++ b/internal/pluginmanager/repository.go @@ -306,6 +306,62 @@ WHERE id = ?`, return err } +func (r Repository) PluginServiceState(ctx context.Context) (PluginServiceState, error) { + row := r.db.QueryRowContext(ctx, ` +SELECT desired_mode, active_mode, applied_at, live_migration, last_error, updated_by, updated_at +FROM plugin_service_state WHERE id = 1`) + state, err := scanPluginServiceState(row) + if errors.Is(err, sql.ErrNoRows) { + now := r.now().Unix() + _, err = r.db.ExecContext(ctx, ` +INSERT OR IGNORE INTO plugin_service_state(id, desired_mode, active_mode, applied_at, live_migration, updated_by, updated_at) +VALUES (1, ?, ?, ?, ?, ?, ?)`, + PluginServiceModeInProcess, PluginServiceModeInProcess, now, PluginMigrationDrainOnly, "system", now) + if err != nil { + return PluginServiceState{}, err + } + return PluginServiceState{ + DesiredMode: PluginServiceModeInProcess, + ActiveMode: PluginServiceModeInProcess, + AppliedAt: now, + LiveMigration: PluginMigrationDrainOnly, + UpdatedBy: "system", + UpdatedAt: now, + }, nil + } + state = normalizePluginServiceState(state) + return state, err +} + +func (r Repository) SetPluginServiceDesired(ctx context.Context, actor, mode string) (PluginServiceState, error) { + now := r.now().Unix() + if _, err := r.db.ExecContext(ctx, ` +INSERT INTO plugin_service_state(id, desired_mode, active_mode, applied_at, live_migration, updated_by, updated_at) +VALUES (1, ?, ?, ?, ?, ?, ?) +ON CONFLICT(id) DO UPDATE SET desired_mode = excluded.desired_mode, updated_by = excluded.updated_by, updated_at = excluded.updated_at`, + mode, PluginServiceModeInProcess, 0, PluginMigrationDrainOnly, actor, now); err != nil { + return PluginServiceState{}, err + } + return r.PluginServiceState(ctx) +} + +func (r Repository) ApplyPluginServiceActive(ctx context.Context, mode string) (PluginServiceState, error) { + now := r.now().Unix() + if _, err := r.db.ExecContext(ctx, ` +INSERT INTO plugin_service_state(id, desired_mode, active_mode, applied_at, live_migration, last_error, updated_by, updated_at) +VALUES (1, ?, ?, ?, ?, '', 'system', ?) +ON CONFLICT(id) DO UPDATE SET active_mode = excluded.active_mode, applied_at = excluded.applied_at, last_error = '', updated_at = excluded.updated_at`, + mode, mode, now, PluginMigrationDrainOnly, now); err != nil { + return PluginServiceState{}, err + } + return r.PluginServiceState(ctx) +} + +func (r Repository) SetPluginServiceError(ctx context.Context, message string) error { + _, err := r.db.ExecContext(ctx, `UPDATE plugin_service_state SET last_error = ?, updated_at = ? WHERE id = 1`, message, r.now().Unix()) + return err +} + func (r Repository) UpdateArtifactStatus(ctx context.Context, artifactID, status, message string) error { _, err := r.db.ExecContext(ctx, `UPDATE plugin_artifacts SET status = ?, error = ?, updated_at = ? WHERE id = ?`, status, message, r.now().Unix(), artifactID) @@ -818,6 +874,151 @@ FROM plugin_advisories` return advisories, rows.Err() } +func (r Repository) SaveRepositoryImport(ctx context.Context, record RepositoryImportRecord) (RepositoryImportRecord, error) { + now := r.now().Unix() + if record.CreatedAt == 0 { + record.CreatedAt = now + } + if record.AdmissionJSON == "" || !json.Valid([]byte(record.AdmissionJSON)) { + record.AdmissionJSON = "{}" + } + _, err := r.db.ExecContext(ctx, ` +INSERT INTO plugin_repository_imports( + repository_type, index_path, repository_name, candidate_id, plugin_id, version, + artifact_id, package_sha256, trust_policy, admission_json, imported_by, created_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + record.RepositoryType, record.IndexPath, record.RepositoryName, record.CandidateID, record.PluginID, record.Version, + record.ArtifactID, record.PackageSHA256, record.TrustPolicy, record.AdmissionJSON, record.ImportedBy, record.CreatedAt) + if err != nil { + return RepositoryImportRecord{}, err + } + id, err := lastInsertID(ctx, r.db) + if err != nil { + return RepositoryImportRecord{}, err + } + return r.RepositoryImport(ctx, id) +} + +func (r Repository) RepositoryImport(ctx context.Context, id int64) (RepositoryImportRecord, error) { + row := r.db.QueryRowContext(ctx, ` +SELECT id, repository_type, index_path, repository_name, candidate_id, plugin_id, version, + artifact_id, package_sha256, trust_policy, admission_json, imported_by, created_at +FROM plugin_repository_imports WHERE id = ?`, id) + return scanRepositoryImport(row) +} + +func (r Repository) ListRepositoryImports(ctx context.Context) ([]RepositoryImportRecord, error) { + rows, err := r.db.QueryContext(ctx, ` +SELECT id, repository_type, index_path, repository_name, candidate_id, plugin_id, version, + artifact_id, package_sha256, trust_policy, admission_json, imported_by, created_at +FROM plugin_repository_imports ORDER BY created_at DESC, id DESC`) + if err != nil { + return nil, err + } + defer rows.Close() + var imports []RepositoryImportRecord + for rows.Next() { + record, err := scanRepositoryImport(rows) + if err != nil { + return nil, err + } + imports = append(imports, record) + } + return imports, rows.Err() +} + +func (r Repository) SaveSupplyChainAssessment(ctx context.Context, assessment SupplyChainAssessment) (SupplyChainAssessment, error) { + now := r.now().Unix() + if assessment.CreatedAt == 0 { + assessment.CreatedAt = now + } + issues, err := json.Marshal(assessment.Issues) + if err != nil { + return SupplyChainAssessment{}, err + } + signature, err := marshalDefaultObject(assessment.Signature) + if err != nil { + return SupplyChainAssessment{}, err + } + sbom, err := marshalDefaultObject(assessment.SBOM) + if err != nil { + return SupplyChainAssessment{}, err + } + license, err := marshalDefaultObject(assessment.License) + if err != nil { + return SupplyChainAssessment{}, err + } + advisory, err := marshalDefaultObject(assessment.Advisory) + if err != nil { + return SupplyChainAssessment{}, err + } + metadata, err := marshalDefaultObject(assessment.Metadata) + if err != nil { + return SupplyChainAssessment{}, err + } + if assessment.Status == "" { + assessment.Status = SupplyChainStatusAllowed + } + _, err = r.db.ExecContext(ctx, ` +INSERT INTO plugin_supply_chain_assessments( + plugin_id, artifact_id, status, issues_json, signature_json, sbom_json, + license_json, advisory_json, metadata_json, created_by, created_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + assessment.PluginID, assessment.ArtifactID, assessment.Status, string(issues), signature, sbom, + license, advisory, metadata, assessment.CreatedBy, assessment.CreatedAt) + if err != nil { + return SupplyChainAssessment{}, err + } + id, err := lastInsertID(ctx, r.db) + if err != nil { + return SupplyChainAssessment{}, err + } + return r.SupplyChainAssessment(ctx, id) +} + +func (r Repository) SupplyChainAssessment(ctx context.Context, id int64) (SupplyChainAssessment, error) { + row := r.db.QueryRowContext(ctx, ` +SELECT id, plugin_id, artifact_id, status, issues_json, signature_json, sbom_json, + license_json, advisory_json, metadata_json, created_by, created_at +FROM plugin_supply_chain_assessments WHERE id = ?`, id) + return scanSupplyChainAssessment(row) +} + +func (r Repository) ListSupplyChainAssessments(ctx context.Context, pluginID, artifactID string) ([]SupplyChainAssessment, error) { + query := ` +SELECT id, plugin_id, artifact_id, status, issues_json, signature_json, sbom_json, + license_json, advisory_json, metadata_json, created_by, created_at +FROM plugin_supply_chain_assessments` + var args []any + var clauses []string + if pluginID != "" { + clauses = append(clauses, "plugin_id = ?") + args = append(args, pluginID) + } + if artifactID != "" { + clauses = append(clauses, "artifact_id = ?") + args = append(args, artifactID) + } + if len(clauses) > 0 { + query += " WHERE " + strings.Join(clauses, " AND ") + } + query += ` ORDER BY created_at DESC, id DESC` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var assessments []SupplyChainAssessment + for rows.Next() { + assessment, err := scanSupplyChainAssessment(rows) + if err != nil { + return nil, err + } + assessments = append(assessments, assessment) + } + return assessments, rows.Err() +} + func (r Repository) SavePreflight(ctx context.Context, record PreflightRecord) (PreflightRecord, error) { now := r.now().Unix() if record.CreatedAt == 0 { @@ -953,6 +1154,72 @@ LIMIT 1`, pluginID, artifactID, profile) return benchmark, err } +func (r Repository) SaveInstrumentation(ctx context.Context, actor string, req InstrumentationRequest) (InstrumentationRecord, error) { + now := r.now().Unix() + if req.Status == "" { + req.Status = InstrumentationStatusAvailable + } + provenance, err := marshalDefaultObject(req.Provenance) + if err != nil { + return InstrumentationRecord{}, err + } + conformance, err := marshalDefaultObject(req.Conformance) + if err != nil { + return InstrumentationRecord{}, err + } + benchmark, err := marshalDefaultObject(req.Benchmark) + if err != nil { + return InstrumentationRecord{}, err + } + smoke, err := marshalDefaultObject(req.Smoke) + if err != nil { + return InstrumentationRecord{}, err + } + _, err = r.db.ExecContext(ctx, ` +INSERT INTO plugin_instrumentation( + name, version, profile, generated_diff_hash, provenance_json, conformance_json, + benchmark_json, smoke_json, runbook_rollback, status, created_by, created_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + req.Name, req.Version, req.Profile, req.GeneratedDiffHash, provenance, conformance, + benchmark, smoke, req.RunbookRollback, req.Status, actor, now) + if err != nil { + return InstrumentationRecord{}, err + } + id, err := lastInsertID(ctx, r.db) + if err != nil { + return InstrumentationRecord{}, err + } + return r.Instrumentation(ctx, id) +} + +func (r Repository) Instrumentation(ctx context.Context, id int64) (InstrumentationRecord, error) { + row := r.db.QueryRowContext(ctx, ` +SELECT id, name, version, profile, generated_diff_hash, provenance_json, conformance_json, + benchmark_json, smoke_json, runbook_rollback, status, created_by, created_at +FROM plugin_instrumentation WHERE id = ?`, id) + return scanInstrumentation(row) +} + +func (r Repository) ListInstrumentation(ctx context.Context) ([]InstrumentationRecord, error) { + rows, err := r.db.QueryContext(ctx, ` +SELECT id, name, version, profile, generated_diff_hash, provenance_json, conformance_json, + benchmark_json, smoke_json, runbook_rollback, status, created_by, created_at +FROM plugin_instrumentation ORDER BY created_at DESC, id DESC`) + if err != nil { + return nil, err + } + defer rows.Close() + var records []InstrumentationRecord + for rows.Next() { + record, err := scanInstrumentation(rows) + if err != nil { + return nil, err + } + records = append(records, record) + } + return records, rows.Err() +} + func (r Repository) RecordOperation(ctx context.Context, pluginID, artifactID, operation, status, actor, message string, metadata any) error { metadataJSON, err := marshalDefaultObject(metadata) if err != nil { @@ -1320,6 +1587,16 @@ type rowScanner interface { Scan(dest ...any) error } +func scanPluginServiceState(row rowScanner) (PluginServiceState, error) { + var state PluginServiceState + err := row.Scan( + &state.DesiredMode, &state.ActiveMode, &state.AppliedAt, &state.LiveMigration, + &state.LastError, &state.UpdatedBy, &state.UpdatedAt, + ) + state = normalizePluginServiceState(state) + return state, err +} + func scanArtifact(row rowScanner) (ArtifactRecord, error) { var artifact ArtifactRecord err := row.Scan( @@ -1427,6 +1704,48 @@ func scanBenchmark(row rowScanner) (BenchmarkRecord, error) { return record, err } +func scanRepositoryImport(row rowScanner) (RepositoryImportRecord, error) { + var record RepositoryImportRecord + err := row.Scan( + &record.ID, &record.RepositoryType, &record.IndexPath, &record.RepositoryName, + &record.CandidateID, &record.PluginID, &record.Version, &record.ArtifactID, + &record.PackageSHA256, &record.TrustPolicy, &record.AdmissionJSON, + &record.ImportedBy, &record.CreatedAt, + ) + return record, err +} + +func scanSupplyChainAssessment(row rowScanner) (SupplyChainAssessment, error) { + var assessment SupplyChainAssessment + var issuesJSON, signatureJSON, sbomJSON, licenseJSON, advisoryJSON, metadataJSON string + err := row.Scan( + &assessment.ID, &assessment.PluginID, &assessment.ArtifactID, &assessment.Status, + &issuesJSON, &signatureJSON, &sbomJSON, &licenseJSON, &advisoryJSON, + &metadataJSON, &assessment.CreatedBy, &assessment.CreatedAt, + ) + if err != nil { + return assessment, err + } + _ = json.Unmarshal([]byte(defaultJSONArray(issuesJSON)), &assessment.Issues) + assessment.Signature = jsonMap(signatureJSON) + assessment.SBOM = jsonMap(sbomJSON) + assessment.License = jsonMap(licenseJSON) + assessment.Advisory = jsonMap(advisoryJSON) + assessment.Metadata = jsonMap(metadataJSON) + return assessment, nil +} + +func scanInstrumentation(row rowScanner) (InstrumentationRecord, error) { + var record InstrumentationRecord + err := row.Scan( + &record.ID, &record.Name, &record.Version, &record.Profile, &record.GeneratedDiffHash, + &record.ProvenanceJSON, &record.ConformanceJSON, &record.BenchmarkJSON, + &record.SmokeJSON, &record.RunbookRollback, &record.Status, &record.CreatedBy, + &record.CreatedAt, + ) + return record, err +} + func marshalDefaultObject(value any) (string, error) { if value == nil { return "{}", nil @@ -1455,6 +1774,28 @@ func defaultJSONArray(value string) string { return value } +func jsonMap(value string) map[string]any { + var out map[string]any + if json.Unmarshal([]byte(defaultJSONObject(value)), &out) != nil { + return nil + } + return out +} + +func normalizePluginServiceState(state PluginServiceState) PluginServiceState { + if state.DesiredMode == "" { + state.DesiredMode = PluginServiceModeInProcess + } + if state.ActiveMode == "" { + state.ActiveMode = PluginServiceModeInProcess + } + if state.LiveMigration == "" { + state.LiveMigration = PluginMigrationDrainOnly + } + state.RestartRequired = state.DesiredMode != state.ActiveMode + return state +} + func boolInt(value bool) int { if value { return 1 diff --git a/internal/pluginmanager/types.go b/internal/pluginmanager/types.go index 5ebe06c..4096253 100644 --- a/internal/pluginmanager/types.go +++ b/internal/pluginmanager/types.go @@ -19,12 +19,17 @@ const ( ArtifactTypeSource = "source" RuntimeGoPlugin = "go-plugin" RuntimeBuiltin = "builtin" + RuntimeSandbox = "sandbox-process" + RuntimeWASM = "wasm" RuntimeEntry = "plugin.so" + RuntimeWASMEntry = "plugin.wasm" SourceBuildEntry = "." ExtensionUpstreamConnect = "upstream.connect/v1" ExtensionRouteResolve = "route.resolve/v1" ExtensionRouteResolver = "route.resolver/v1" + ExtensionRuleEvaluate = "rule.evaluate/v1" + ExtensionConfigValidate = "config.validate/v1" ExtensionStatusPing = "status.ping/v1" ExtensionConnectionFilter = "connection.filter/v1" ExtensionHandshakeFilter = "handshake.filter/v1" @@ -64,6 +69,26 @@ const ( RuntimeDisabled = "disabled" RuntimeDraining = "draining" + PluginServiceModeInProcess = "in-process" + PluginServiceModeGoPluginProcess = "go-plugin-process" + PluginServiceModeSandboxProcess = "sandbox-process" + + PluginMigrationDrainOnly = "drain-only" + PluginMigrationFDLive = "fd-live" + PluginMigrationFDLiveSHM = "fd-live-shm" + + RepositoryTypeOfficial = "official" + RepositoryTypeInternal = "internal" + RepositoryTypeFile = "file" + RepositoryTypeURL = "url" + + SupplyChainStatusAllowed = "allowed" + SupplyChainStatusBlocked = "blocked" + SupplyChainStatusWarning = "warning" + + InstrumentationStatusAvailable = "available" + InstrumentationStatusBlocked = "blocked" + PolicyProfileDev = "dev" PolicyProfileStaging = "staging" PolicyProfileProd = "prod" @@ -173,6 +198,7 @@ type ExtensionPoint struct { type RuntimeLimits struct { HandlerTimeoutMS int `json:"handler_timeout_ms"` InitialWriteTimeoutMS int `json:"initial_write_timeout_ms"` + MemoryBytes int `json:"memory_bytes,omitempty"` } type SecretSpec struct { @@ -254,9 +280,15 @@ type CapabilitySummary struct { ExternalDeps []ExternalSpec `json:"external_dependencies,omitempty"` DataStores []DataStoreSpec `json:"data_stores,omitempty"` FileStores []FileStoreSpec `json:"file_stores,omitempty"` + Runtime RuntimeCapability `json:"runtime,omitempty"` Raw json.RawMessage `json:"raw,omitempty"` } +type RuntimeCapability struct { + RequiredCapabilities []string `json:"required_capabilities,omitempty"` + RequiredFeatures []string `json:"required_features,omitempty"` +} + type UpstreamConnectCapability struct { Mode string `json:"mode,omitempty"` } @@ -670,6 +702,105 @@ type BuildRequest struct { VendorRequired bool `json:"vendor_required"` } +type PluginServiceState struct { + DesiredMode string `json:"desired_mode"` + ActiveMode string `json:"active_mode"` + AppliedAt int64 `json:"applied_at"` + RestartRequired bool `json:"restart_required"` + LiveMigration string `json:"live_migration"` + LastError string `json:"last_error"` + UpdatedBy string `json:"updated_by"` + UpdatedAt int64 `json:"updated_at"` +} + +type PluginServiceStatus struct { + Service PluginServiceState `json:"service"` + Hosts []PluginHostRuntimeSummary `json:"hosts"` +} + +type PluginHostRuntimeSummary struct { + PluginID string `json:"plugin_id"` + ArtifactID string `json:"artifact_id"` + State string `json:"state"` + DrainMode string `json:"drain_mode"` + CrashLoop bool `json:"crash_loop"` + CrashCount int `json:"crash_count"` + LastError string `json:"last_error"` + StartedAt int64 `json:"started_at"` + DrainingAt int64 `json:"draining_at"` + ExitedAt int64 `json:"exited_at"` + LastCrashAt int64 `json:"last_crash_at"` +} + +type RepositoryImportRequest struct { + RepositoryType string `json:"repository_type"` + IndexPath string `json:"index_path"` + ArtifactID string `json:"artifact_id"` + PluginID string `json:"plugin_id"` + Version string `json:"version"` + TrustPolicy string `json:"trust_policy"` +} + +type RepositoryImportRecord struct { + ID int64 `json:"id"` + RepositoryType string `json:"repository_type"` + IndexPath string `json:"index_path"` + RepositoryName string `json:"repository_name"` + CandidateID string `json:"candidate_id"` + PluginID string `json:"plugin_id"` + Version string `json:"version"` + ArtifactID string `json:"artifact_id"` + PackageSHA256 string `json:"package_sha256"` + TrustPolicy string `json:"trust_policy"` + AdmissionJSON string `json:"admission_json"` + ImportedBy string `json:"imported_by"` + CreatedAt int64 `json:"created_at"` +} + +type SupplyChainAssessment struct { + ID int64 `json:"id"` + PluginID string `json:"plugin_id"` + ArtifactID string `json:"artifact_id"` + Status string `json:"status"` + Issues []GovernanceIssue `json:"issues"` + Signature map[string]any `json:"signature,omitempty"` + SBOM map[string]any `json:"sbom,omitempty"` + License map[string]any `json:"license,omitempty"` + Advisory map[string]any `json:"advisory,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + CreatedBy string `json:"created_by"` + CreatedAt int64 `json:"created_at"` +} + +type InstrumentationRecord struct { + ID int64 `json:"id"` + Name string `json:"name"` + Version string `json:"version"` + Profile string `json:"profile"` + GeneratedDiffHash string `json:"generated_diff_hash"` + ProvenanceJSON string `json:"provenance_json"` + ConformanceJSON string `json:"conformance_json"` + BenchmarkJSON string `json:"benchmark_json"` + SmokeJSON string `json:"smoke_json"` + RunbookRollback string `json:"runbook_rollback"` + Status string `json:"status"` + CreatedBy string `json:"created_by"` + CreatedAt int64 `json:"created_at"` +} + +type InstrumentationRequest struct { + Name string `json:"name"` + Version string `json:"version"` + Profile string `json:"profile"` + GeneratedDiffHash string `json:"generated_diff_hash"` + Provenance map[string]any `json:"provenance"` + Conformance map[string]any `json:"conformance"` + Benchmark map[string]any `json:"benchmark"` + Smoke map[string]any `json:"smoke"` + RunbookRollback string `json:"runbook_rollback"` + Status string `json:"status"` +} + type GCCandidate struct { Kind string `json:"kind"` ID string `json:"id"`