From dd33976faf50e112e1a2c9bbded091bdebc2c9cc Mon Sep 17 00:00:00 2001 From: Victor Lyuboslavsky <2685025+getvictor@users.noreply.github.com> Date: Fri, 5 Jun 2026 01:13:03 -0500 Subject: [PATCH] osquery_perf: Windows MDM push (#46777) **Related issue:** Resolves #46567 Note: Hide whitespace for better review ## Testing - [x] QA'd all new/changed functionality manually ## Summary by CodeRabbit * **New Features** * Server-triggered on-demand Windows MDM check-ins for immediate device syncs * Dynamic adjustment of the device polling interval based on server directives * Enhanced metrics: tracking and reporting of on-demand MDM synchronization sessions --- cmd/osquery-perf/agent.go | 213 +++++++++++++++++-------- cmd/osquery-perf/osquery_perf/stats.go | 12 +- 2 files changed, 154 insertions(+), 71 deletions(-) diff --git a/cmd/osquery-perf/agent.go b/cmd/osquery-perf/agent.go index ff00d3ed23..52e45c3bf2 100644 --- a/cmd/osquery-perf/agent.go +++ b/cmd/osquery-perf/agent.go @@ -463,6 +463,11 @@ type agent struct { macMDMClient *mdmtest.TestAppleMDMClient winMDMClient *mdmtest.TestWindowsMDMClient + // winMDMWake signals the Windows MDM loop to start an OMA-DM session on demand (in response to the server's + // WindowsMDMSyncRequest notification), mirroring real fleetd waking the device. Buffered with capacity 1, sent + // non-blocking, so coalesced wakes never block the orbit config loop. Non-nil only for Windows MDM agents. + winMDMWake chan struct{} + // isEnrolledToMDM is true when the mdmDevice has enrolled. isEnrolledToMDM bool // isEnrolledToMDMMu protects isEnrolledToMDM. @@ -729,6 +734,11 @@ func newAgent( entraIDUserPrincipalName: fmt.Sprintf("fake-%s@example.com", randomString(5)), } + // Windows MDM agents can be woken on demand by the server, so give them a wake channel for the MDM loop. + if winMDMClient != nil { + agent.winMDMWake = make(chan struct{}, 1) + } + // Initialize host identity client agent.hostIdentityClient = hostidentity.NewClient(hostidentity.Config{ ServerAddress: serverAddress, @@ -1013,6 +1023,16 @@ func (a *agent) runOrbitLoop() { orbitClient.TestNodeKey = *a.orbitNodeKey + // Simulated Windows MDM hosts advertise CapabilityWindowsMDMSync, like real Windows fleetd, so the server relaxes + // their DMClient poll schedule (via a Replace on the poll node) and wakes them on demand via WindowsMDMSyncRequest + // instead of relying on frequent polling. Real fleetd adds this at construction (GetOrbitClientCapabilities gates it + // on GOOS=windows); osquery-perf simulates Windows on a non-Windows GOOS, so we add it here. Mutating the map directly + // is safe: GetOrbitClientCapabilities returns a fresh per-client map (not shared), and this runs during setup before + // the first request or any concurrent goroutine, so there is no copy needed. + if a.winMDMClient != nil { + orbitClient.ClientCapabilities[fleet.CapabilityWindowsMDMSync] = struct{}{} + } + deviceClient, err := fleetclient.NewDeviceClient(a.serverAddress, true, "", nil, "") if err != nil { log.Fatal("creating device client: ", err) @@ -1121,6 +1141,14 @@ func (a *agent) runOrbitLoop() { go a.runWindowsMDMLoop() } } + if cfg.Notifications.WindowsMDMSyncRequest && a.mdmEnrolled() && a.winMDMWake != nil { + // The server has queued Windows MDM commands and asked this (relaxed-poll) host to start an OMA-DM + // session now. + select { + case a.winMDMWake <- struct{}{}: + default: + } + } case <-orbitTokenRemoteCheckTicker: if !a.disableFleetDesktop && tokenRotationEnabled { if err := deviceClient.CheckToken(*a.deviceAuthToken); err != nil { @@ -1526,83 +1554,128 @@ func (a *agent) ddmSendStatus(items *fleet.MDMAppleDDMDeclarationItemsResponse) } func (a *agent) runWindowsMDMLoop() { - mdmCheckInTicker := time.Tick(a.MDMCheckInInterval) + pollInterval := a.MDMCheckInInterval + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() - for range mdmCheckInTicker { - cmds, err := a.winMDMClient.StartManagementSession() - if err != nil { - log.Printf("MDM check-in start session request failed: %s", err) - a.stats.IncrementMDMErrors() - continue - } - a.stats.IncrementMDMSessions() - - // send a successful ack for each command - msgID, err := a.winMDMClient.GetCurrentMsgID() - if err != nil { - log.Printf("MDM get current MsgID failed: %s", err) - a.stats.IncrementMDMErrors() - continue - } - - // Detect SCEP CertificateInstall CSPs and ACK them now while kicking off the SCEP exchange in the background. - scepCtx, cancelSCEP := context.WithTimeout(context.Background(), 2*a.MDMCheckInInterval) - handled, scepResults, hasWork := a.winMDMClient.AppendSCEPInstallResponses(scepCtx, cmds, msgID, nil) - if hasWork { - go func() { - defer cancelSCEP() - // One SCEPResult is emitted per CSP; increment per-result so the request counter - // matches the per-CSP success/error counters even when multiple CSPs ride one SyncML. - for res := range scepResults { - a.stats.IncrementMDMSCEPRequests() - if res.Err != nil { - log.Printf("MDM SCEP exchange failed: %s", res.Err) - a.stats.IncrementMDMSCEPErrors() - continue - } - a.stats.IncrementMDMSCEPSuccess() - } - }() - } else { - cancelSCEP() - } - - for _, c := range cmds { - // Skip the server's own entries. MS-MDM's "Status on a Status" is only for auth-renegotiation edge cases (see - // https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-mdm/36b1a4d9-fd93-48ce-b865-6a9d396c52a4 - // "While this case is not usually encountered"); real Windows does not emit Status-on-Status during normal check-ins. - if c.Verb == fleet.CmdStatus { - continue + for { + onDemand := false + select { + case <-a.winMDMWake: + // The server asked us (via WindowsMDMSyncRequest) to start a session now, without waiting for the poll. + onDemand = true + case <-ticker.C: + // A wake may have arrived at almost the same time as this tick; select picks at random when both are ready, + // which could miscount a wake-triggered session as poll-triggered. Drain any buffered wake first so the + // on-demand metric is deterministic (and a coincident tick+wake collapses into a single on-demand session). + select { + case <-a.winMDMWake: + onDemand = true + default: } - if _, ok := handled[c.Cmd.CmdID.Value]; ok { - // Already ACKed by AppendSCEPInstallResponses. - a.stats.IncrementMDMCommandsReceived() - continue - } - a.stats.IncrementMDMCommandsReceived() - - status := syncml.CmdStatusOK - if a.mdmProfileFailureProb > 0.0 && rand.Float64() <= a.mdmProfileFailureProb { - status = syncml.CmdStatusBadRequest - } - a.winMDMClient.AppendResponse(fleet.SyncMLCmd{ - XMLName: xml.Name{Local: fleet.CmdStatus}, - MsgRef: &msgID, - CmdRef: &c.Cmd.CmdID.Value, - Cmd: ptr.String(c.Verb), - Data: &status, - Items: nil, - CmdID: fleet.CmdID{Value: uuid.NewString()}, - }) } - if _, err := a.winMDMClient.SendResponse(); err != nil { - log.Printf("MDM send response request failed: %s", err) - a.stats.IncrementMDMErrors() - continue + + // If the server relaxed (or restored) our DMClient poll schedule via a Replace on the poll node, match it so our + // scheduled polling slows down. Steady-state command delivery is then driven by the on-demand wake rather than + // frequent polling, which is the behavior this load test exercises. + if relaxedInterval := a.doWindowsMDMCheckIn(onDemand); relaxedInterval > 0 && relaxedInterval != pollInterval { + pollInterval = relaxedInterval + ticker.Reset(pollInterval) + log.Printf("host %d: Windows MDM poll schedule set to %s", a.agentIndex, pollInterval) } } } +// doWindowsMDMCheckIn runs a single OMA-DM management session: it starts the session, acknowledges every command the +// server sends, and returns the new scheduled poll interval if the server sent a Replace on the DMClient poll node +// (0 otherwise). onDemand reports whether the session was triggered by a WindowsMDMSyncRequest wake instead of the poll +// ticker, for stats purposes. +func (a *agent) doWindowsMDMCheckIn(onDemand bool) (newPollInterval time.Duration) { + cmds, err := a.winMDMClient.StartManagementSession() + if err != nil { + log.Printf("MDM check-in start session request failed: %s", err) + a.stats.IncrementMDMErrors() + return 0 + } + a.stats.IncrementMDMSessions() + if onDemand { + a.stats.IncrementMDMOnDemandSyncs() + } + + // send a successful ack for each command + msgID, err := a.winMDMClient.GetCurrentMsgID() + if err != nil { + log.Printf("MDM get current MsgID failed: %s", err) + a.stats.IncrementMDMErrors() + return 0 + } + + // Detect SCEP CertificateInstall CSPs and ACK them now while kicking off the SCEP exchange in the background. + scepCtx, cancelSCEP := context.WithTimeout(context.Background(), 2*a.MDMCheckInInterval) + handled, scepResults, hasWork := a.winMDMClient.AppendSCEPInstallResponses(scepCtx, cmds, msgID, nil) + if hasWork { + go func() { + defer cancelSCEP() + // One SCEPResult is emitted per CSP; increment per-result so the request counter + // matches the per-CSP success/error counters even when multiple CSPs ride one SyncML. + for res := range scepResults { + a.stats.IncrementMDMSCEPRequests() + if res.Err != nil { + log.Printf("MDM SCEP exchange failed: %s", res.Err) + a.stats.IncrementMDMSCEPErrors() + continue + } + a.stats.IncrementMDMSCEPSuccess() + } + }() + } else { + cancelSCEP() + } + + for _, c := range cmds { + // Skip the server's own entries. MS-MDM's "Status on a Status" is only for auth-renegotiation edge cases (see + // https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-mdm/36b1a4d9-fd93-48ce-b865-6a9d396c52a4 + // "While this case is not usually encountered"); real Windows does not emit Status-on-Status during normal check-ins. + if c.Verb == fleet.CmdStatus { + continue + } + // When the server relaxes (or restores) our poll schedule it sends a Replace on the DMClient poll node. Honor it + // by adjusting our scheduled poll interval, mirroring how the real Windows DMClient applies the Replace. We still + // ACK it below like any other command. + if c.Verb == fleet.CmdReplace && c.Cmd.GetTargetURI() == syncml.DMClientPollIntervalLocURI { + if mins, err := strconv.Atoi(strings.TrimSpace(c.Cmd.GetTargetData())); err == nil && mins > 0 { + newPollInterval = time.Duration(mins) * time.Minute + } + } + if _, ok := handled[c.Cmd.CmdID.Value]; ok { + // Already ACKed by AppendSCEPInstallResponses. + a.stats.IncrementMDMCommandsReceived() + continue + } + a.stats.IncrementMDMCommandsReceived() + + status := syncml.CmdStatusOK + if a.mdmProfileFailureProb > 0.0 && rand.Float64() <= a.mdmProfileFailureProb { + status = syncml.CmdStatusBadRequest + } + a.winMDMClient.AppendResponse(fleet.SyncMLCmd{ + XMLName: xml.Name{Local: fleet.CmdStatus}, + MsgRef: &msgID, + CmdRef: &c.Cmd.CmdID.Value, + Cmd: ptr.String(c.Verb), + Data: &status, + Items: nil, + CmdID: fleet.CmdID{Value: uuid.NewString()}, + }) + } + if _, err := a.winMDMClient.SendResponse(); err != nil { + log.Printf("MDM send response request failed: %s", err) + a.stats.IncrementMDMErrors() + return 0 + } + return newPollInterval +} + func (a *agent) execScripts(execIDs []string, orbitClient *fleetclient.OrbitClient) { if a.scriptExecRunning.Swap(true) { // if Swap returns true, the goroutine was already running, exit diff --git a/cmd/osquery-perf/osquery_perf/stats.go b/cmd/osquery-perf/osquery_perf/stats.go index 7aeee0fd09..05c3157c87 100644 --- a/cmd/osquery-perf/osquery_perf/stats.go +++ b/cmd/osquery-perf/osquery_perf/stats.go @@ -13,6 +13,7 @@ type Stats struct { orbitEnrollments int mdmEnrollments int mdmSessions int + mdmOnDemandSyncs int distributedWrites int mdmCommandsReceived int mdmSCEPRequests int @@ -77,6 +78,14 @@ func (s *Stats) IncrementMDMSessions() { s.mdmSessions++ } +// IncrementMDMOnDemandSyncs counts Windows MDM sessions that were triggered by an on-demand wake +// (WindowsMDMSyncRequest) rather than the poll ticker. This is a subset of mdmSessions, not a separate total. +func (s *Stats) IncrementMDMOnDemandSyncs() { + s.l.Lock() + defer s.l.Unlock() + s.mdmOnDemandSyncs++ +} + func (s *Stats) IncrementDistributedWrites() { s.l.Lock() defer s.l.Unlock() @@ -265,7 +274,7 @@ func (s *Stats) Log() { defer s.l.Unlock() log.Printf( - "uptime: %s, error rate: %.2f, osquery enrolls: %d, orbit enrolls: %d, mdm enrolls: %d, distributed/reads: %d, distributed/writes: %d, config requests: %d, result log requests: %d, mdm sessions initiated: %d, mdm commands received: %d, config errors: %d, distributed/read errors: %d, distributed/write errors: %d, log result errors: %d, orbit errors: %d, desktop errors: %d, mdm errors: %d, mdm scep requests: %d, mdm scep success: %d, mdm scep errors: %d, ddm tokens success: %d, ddm tokens errors: %d, ddm declaration items success: %d, ddm declaration items errors: %d, ddm activation success: %d, ddm activation errors: %d, ddm configuration success: %d, ddm configuration errors: %d, ddm status success: %d, ddm status errors: %d, buffered logs: %d, script execs (errs): %d (%d), software installs (errs): %d (%d)", + "uptime: %s, error rate: %.2f, osquery enrolls: %d, orbit enrolls: %d, mdm enrolls: %d, distributed/reads: %d, distributed/writes: %d, config requests: %d, result log requests: %d, mdm sessions initiated: %d, mdm on-demand syncs: %d, mdm commands received: %d, config errors: %d, distributed/read errors: %d, distributed/write errors: %d, log result errors: %d, orbit errors: %d, desktop errors: %d, mdm errors: %d, mdm scep requests: %d, mdm scep success: %d, mdm scep errors: %d, ddm tokens success: %d, ddm tokens errors: %d, ddm declaration items success: %d, ddm declaration items errors: %d, ddm activation success: %d, ddm activation errors: %d, ddm configuration success: %d, ddm configuration errors: %d, ddm status success: %d, ddm status errors: %d, buffered logs: %d, script execs (errs): %d (%d), software installs (errs): %d (%d)", time.Since(s.StartTime).Round(time.Second), float64(s.errors)/float64(s.osqueryEnrollments), s.osqueryEnrollments, @@ -276,6 +285,7 @@ func (s *Stats) Log() { s.configRequests, s.resultLogRequests, s.mdmSessions, + s.mdmOnDemandSyncs, s.mdmCommandsReceived, s.configErrors, s.distributedReadErrors,