admin343/browser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tighten opaque plugin content boundaries

Browse Source
This commit is contained in:
npc0-hue
2026-09-03 18:24:39 +08:00
parent 80cddbf19d
commit 14cbc63e61
31 changed files with 452 additions and 558 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/api/log_events_handlers.go
+31 -47
View File
@@ -17,7 +17,6 @@ type componentLogServerContextKey struct{}
const (
logEventHeartbeatInterval = 15 * time.Second
managedLogSessionIDPrefix = "log-session:"
)
// serverLogEvents streams platform-accepted live append events for the terminal drawer.
@@ -26,7 +25,7 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
writeMethodNotAllowed(w, http.MethodGet)
return
}
instance, streams, liveEligible, subscription, err := h.openLogEventSubscription(r)
instance, streams, subscription, err := h.openLogEventSubscription(r)
if err != nil {
writeServiceError(w, err)
return
@@ -46,12 +45,10 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
active := supervisedLogSession{}
if liveEligible {
if isComponentLogRequest(r) {
active = activeComponentLogSession(streams)
} else {
active = activeSupervisedLogSession(streams)
}
if isComponentLogRequest(r) {
active = activeComponentLogSession(streams)
} else {
active = activeSupervisedLogSession(streams)
}
emittedThrough, err := h.writeCurrentLogSession(w, instance.ID, active)
if err != nil {
@@ -82,7 +79,6 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
continue
}
if subscriptionEvent.ProcessState != domain.ServerInstanceStateRunning {
liveEligible = false
if active.sessionID == "" {
continue
}
@@ -94,14 +90,11 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
flusher.Flush()
continue
}
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
streams, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
if err != nil {
return
}
next := supervisedLogSession{}
if liveEligible {
next = activeSupervisedLogSession(streams)
}
next := activeSupervisedLogSession(streams)
if sameSupervisedLogSession(active, next) {
continue
}
@@ -113,7 +106,7 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
flusher.Flush()
continue
}
if subscriptionEvent.Kind != service.LogEventSubscriptionEventLog || !liveEligible {
if subscriptionEvent.Kind != service.LogEventSubscriptionEventLog {
continue
}
event := subscriptionEvent.LogEvent
@@ -122,20 +115,16 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
candidate = supervisedLogSession{}
}
if candidate.sessionID != "" && newerLogSession(candidate, active) {
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
if err != nil {
return
}
next := supervisedLogSession{}
if liveEligible {
next = activeSupervisedLogSession(streams)
}
next := candidate
if !sameSupervisedLogSession(active, next) {
active = next
emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active)
if err != nil {
return
}
if event.Entry.Seq > 0 && emittedThrough[event.Stream.ID] >= event.Entry.Seq {
emittedThrough[event.Stream.ID] = event.Entry.Seq - 1
}
flusher.Flush()
}
}
@@ -251,10 +240,9 @@ func (h *coreHandlers) writeCurrentLogSession(w http.ResponseWriter, serverInsta
return emittedThrough, nil
}
func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerInstance, []domain.LogStream, bool, service.LogEventSubscription, error) {
func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerInstance, []domain.LogStream, service.LogEventSubscription, error) {
var instance domain.ServerInstance
var streams []domain.LogStream
var liveEligible bool
var subscription service.LogEventSubscription
var err error
if serverInstanceID, ok := r.Context().Value(componentLogServerContextKey{}).(string); ok && strings.TrimSpace(serverInstanceID) != "" {
@@ -266,29 +254,29 @@ func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerI
sessionID := bearerToken(r)
instance, err = h.core.GetServerInstanceForSession(sessionID, r.PathValue("id"))
if err != nil {
return domain.ServerInstance{}, nil, false, subscription, err
return domain.ServerInstance{}, nil, subscription, err
}
subscription, err = h.core.SubscribeLogEventsForSession(sessionID, instance.ID)
if err != nil {
return domain.ServerInstance{}, nil, false, subscription, err
return domain.ServerInstance{}, nil, subscription, err
}
} else {
instance, err = h.core.GetServerInstance(r.PathValue("id"))
if err != nil {
return domain.ServerInstance{}, nil, false, subscription, err
return domain.ServerInstance{}, nil, subscription, err
}
subscription, err = h.core.SubscribeLogEvents(instance.ID)
if err != nil {
return domain.ServerInstance{}, nil, false, subscription, err
return domain.ServerInstance{}, nil, subscription, err
}
}
if err == nil {
streams, liveEligible, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
streams, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r))
}
if err != nil && subscription.Close != nil {
subscription.Close()
}
return instance, streams, liveEligible, subscription, err
return instance, streams, subscription, err
}
func withComponentLogServer(r *http.Request, serverInstanceID string) *http.Request {
@@ -300,46 +288,42 @@ func isComponentLogRequest(r *http.Request) bool {
return ok
}
func (h *coreHandlers) loadLiveLogSnapshot(serverInstanceID string, includeDeclaredStreams bool) ([]domain.LogStream, bool, error) {
func (h *coreHandlers) loadLiveLogSnapshot(serverInstanceID string, includeDeclaredStreams bool) ([]domain.LogStream, error) {
instance, err := h.core.GetServerInstance(serverInstanceID)
if err != nil {
return nil, false, err
return nil, err
}
if instance.State != domain.ServerInstanceStateRunning || strings.TrimSpace(instance.RunEndpointID) == "" {
return nil, false, nil
if strings.TrimSpace(instance.RunEndpointID) == "" {
return nil, nil
}
if !includeDeclaredStreams && instance.State != domain.ServerInstanceStateRunning {
return nil, nil
}
endpoint, err := h.core.GetRunEndpoint(instance.RunEndpointID)
if err != nil {
return nil, false, err
return nil, err
}
if endpoint.Status != domain.RunEndpointStatusOnline {
return nil, false, nil
}
logSessionID := strings.TrimPrefix(instance.LifecycleProcessID, managedLogSessionIDPrefix)
if logSessionID == instance.LifecycleProcessID || strings.TrimSpace(logSessionID) == "" {
return nil, false, nil
return nil, nil
}
streams, err := h.core.ListLogStreams(domain.LogStreamFilter{ServerInstanceID: instance.ID})
if err != nil {
return nil, false, err
return nil, err
}
current := make([]domain.LogStream, 0, len(streams))
for _, stream := range streams {
if includeDeclaredStreams {
if stream.LogSessionID != "" && stream.LogSessionID != logSessionID {
continue
}
if stream.Source != domain.LogStreamSourceProcess && stream.Source != domain.LogStreamSourceFile && stream.Source != domain.LogStreamSourceManagementProgram {
continue
}
current = append(current, stream)
continue
}
if stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == logSessionID {
if stream.Source == domain.LogStreamSourceProcess && strings.TrimSpace(stream.LogSessionID) != "" && !stream.SessionStartedAt.IsZero() {
current = append(current, stream)
}
}
return current, true, nil
return current, nil
}
func writeSSEJSON(w http.ResponseWriter, eventName string, id string, value any) error {
platform/api/log_ingest_handlers_test.go
+10 -6
View File
@@ -207,7 +207,7 @@ func TestLogEventsSSEExcludesOlderSupervisedSessions(t *testing.T) {
}
}
func TestLogEventsSSEInitialSnapshotRequiresRunningOnlineProcessFact(t *testing.T) {
func TestLogEventsSSEInitialSnapshotUsesRunStreamEnvelopeWhenOnline(t *testing.T) {
router := newTestRouter()
hello := createLogIngestAPIFixtures(t, router)
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequest(t, hello.SessionToken, 1, 1)), http.StatusOK)
@@ -222,7 +222,9 @@ func TestLogEventsSSEInitialSnapshotRequiresRunningOnlineProcessFact(t *testing.
running.ObservedAt = time.Date(2026, 7, 3, 13, 0, 0, 0, time.UTC)
running.ExecutionResult = dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "running"}
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", running), http.StatusOK)
assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events"))
if body := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events").Body.String(); !strings.Contains(body, `"logSessionId":"session-current"`) || strings.Contains(body, `"logSessionId":"session-missing"`) {
t.Fatalf("expected initial session to come from Run log stream envelope, body=%s", body)
}
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/control/heartbeat", dto.RunControlHeartbeatRequest{RunEndpointID: "run-local", SessionToken: hello.SessionToken, Version: "0.1.0", Status: domain.RunEndpointStatusOffline, CapabilityFingerprint: "cap-logs", Capacity: dto.RunCapacityResponse{MaxJobs: 1}}), http.StatusOK)
assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events"))
}
@@ -266,14 +268,15 @@ func TestLogEventsSSEClearsStoppedSessionAndRestoresRunningSessionWithoutReconne
next.LogSessionID = "session-next"
next.SessionStartedAt = nextStartedAt
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", next), http.StatusOK)
assertSSEEvent(t, reader, "session", `"logSessionId":"session-next"`)
assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stdout"`)
assertSSEEvent(t, reader, "log", `"streamId":"run.run-local.server-1.session-next.stdout"`)
statusReport.ManagedProcessID = "log-session:session-next"
statusReport.ObservationSeq = 1
statusReport.ObservedAt = nextStartedAt
statusReport.ExecutionResult = dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "running"}
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", statusReport), http.StatusOK)
assertSSEEvent(t, reader, "session", `"logSessionId":"session-next"`)
assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stdout"`)
nextLive := validLogBatchRequestForStream(t, hello.SessionToken, next.LogStreamID, "stdout", 2, 2, 21)
nextLive.LogSessionID = "session-next"
nextLive.SessionStartedAt = nextStartedAt
@@ -327,11 +330,12 @@ func TestLogEventsSSEOrdersSessionSwitchAndAdditionalStreamWithoutDuplicates(t *
if _, err := core.IngestLogBatch(next.ToDomain()); err != nil {
t.Fatalf("ingest next session stdout: %v", err)
}
assertSSEEvent(t, reader, "session", `"logSessionId":"session-next"`)
assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stdout"`)
assertSSEEvent(t, reader, "log", `"streamId":"run.run-local.server-1.session-next.stdout"`)
if _, err := core.ReportRunLifecycle(domain.RunLifecycleReport{RunEndpointID: "run-local", SessionToken: hello.SessionToken, ServerInstanceID: "server-1", Capability: domain.LifecycleCapabilityStatus, State: domain.JobStateSucceeded, ManagedProcessID: "log-session:session-next", ObservationSeq: 1, ObservedAt: nextStartedAt, ExecutionResult: domain.JobExecutionResult{Kind: "process", ProcessState: "running"}}); err != nil {
t.Fatalf("report next managed process: %v", err)
}
assertSSEEvent(t, reader, "session", `"logSessionId":"session-next"`)
assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stdout"`)
stderr := validLogBatchRequestForStream(t, hello.SessionToken, "run.run-local.server-1.session-next.stderr", "stderr", 1, 1, 21)
stderr.LogSessionID = "session-next"
platform/api/resource_handlers_test.go
+5 -3
View File
@@ -1398,7 +1398,7 @@ func TestPluginBridgeExecuteAPI(t *testing.T) {
t.Fatalf("expected permission denied safe envelope, got %+v", denied)
}
unsafe := requestJSONWithAuth(t, router, http.MethodPost, "/api/v1/plugin-bridge/execute", dto.PluginBridgeExecuteRequest{
unsafe := postOKJSONWithAuth[dto.PluginBridgeExecuteResponse](t, router, "/api/v1/plugin-bridge/execute", dto.PluginBridgeExecuteRequest{
RequestID: "bridge-unsafe-1",
PluginID: "game.example",
RouteKey: "logs",
@@ -1406,9 +1406,11 @@ func TestPluginBridgeExecuteAPI(t *testing.T) {
Action: string(domain.PluginBridgeActionFilesRequest),
Payload: map[string]string{"key": "/Users/tasia/.ssh/id_rsa", "idempotencyKey": "idem-unsafe"},
}, ownerSession)
assertErrorResponse(t, unsafe, http.StatusBadRequest, errorCodeValidation)
if unsafe.Status != "error" || unsafe.Error == nil || unsafe.Error.Code != "execution_failed" {
t.Fatalf("expected scoped file validation envelope, got %+v", unsafe)
}
for _, body := range []string{mustJSON(t, serverRead), mustJSON(t, logs), mustJSON(t, lifecycle), mustJSON(t, lifecycleMismatch), mustJSON(t, fileDispatch), mustJSON(t, aiResponse), mustJSON(t, denied)} {
for _, body := range []string{mustJSON(t, serverRead), mustJSON(t, logs), mustJSON(t, lifecycle), mustJSON(t, lifecycleMismatch), mustJSON(t, fileDispatch), mustJSON(t, aiResponse), mustJSON(t, denied), mustJSON(t, unsafe)} {
for _, forbidden := range []string{"/Users/", "unix://", "Bearer ", "sk-", "password=", "apiKeyRef", "rawApiKey"} {
if strings.Contains(body, forbidden) {
t.Fatalf("bridge response exposed forbidden fragment %q: %s", forbidden, body)