diff --git a/agent/ShopDbEndpointAgent.Core/CommandExecutor.cs b/agent/ShopDbEndpointAgent.Core/CommandExecutor.cs index 7db7a198..6bd3315b 100644 --- a/agent/ShopDbEndpointAgent.Core/CommandExecutor.cs +++ b/agent/ShopDbEndpointAgent.Core/CommandExecutor.cs @@ -144,6 +144,14 @@ public sealed class CommandExecutor Message = "Fresh telemetry collection requested", }; + // Full TCP connection table, on demand only. Deliberately not part of + // the standard report: established connections churn every second, so + // sending them on cadence would bloat every payload and swamp the + // stable listener list. As a command the technician pays that cost + // once, at the moment they are actually diagnosing connectivity. + case "network_connections": + return NetworkConnections(); + default: logger.LogWarning("Rejected unknown command action {Action}", action); return new CommandResult @@ -154,6 +162,98 @@ public sealed class CommandExecutor } } + // Snapshot of every TCP connection with its owning process. Capped because a + // busy machine can hold thousands of sockets and an unbounded result would + // be rejected by the server's payload limit, returning nothing at all - + // a truncated answer is far more useful than a failed one. + private const int MaxReportedConnections = 500; + + private CommandResult NetworkConnections() + { + try + { + var rows = TcpTableReader.GetTcpTable(); + var names = GetProcessNameSnapshot(); + var connections = new List(); + var truncated = false; + + foreach (var row in rows) + { + if (connections.Count >= MaxReportedConnections) + { + truncated = true; + break; + } + + names.TryGetValue(row.ProcessId, out var processName); + connections.Add(new Dictionary + { + ["localaddress"] = row.LocalAddress, + ["localport"] = row.LocalPort, + ["remoteaddress"] = row.RemoteAddress, + ["remoteport"] = row.RemotePort, + ["state"] = row.State.ToString().ToLowerInvariant(), + ["processid"] = row.ProcessId, + ["processname"] = processName, + }); + } + + logger.LogInformation( + "network_connections returned {Count} connections (truncated: {Truncated})", + connections.Count, truncated); + + return new CommandResult + { + Succeeded = true, + Message = $"Collected {connections.Count} TCP connections" + + (truncated ? " (truncated)" : ""), + Detail = new Dictionary + { + ["connections"] = connections, + ["totalcount"] = rows.Count, + ["truncated"] = truncated, + }, + }; + } + catch (Exception exception) + { + logger.LogWarning(exception, "network_connections failed"); + return new CommandResult + { + Succeeded = false, + Message = "Could not read the TCP connection table: " + exception.Message, + }; + } + } + + private Dictionary GetProcessNameSnapshot() + { + var names = new Dictionary(); + try + { + foreach (var process in Process.GetProcesses()) + { + try + { + names[process.Id] = process.ProcessName; + } + catch + { + // The process exited between enumeration and read. + } + finally + { + process.Dispose(); + } + } + } + catch (Exception exception) + { + logger.LogDebug(exception, "Process name snapshot failed"); + } + return names; + } + // Guard: a generic action with no target has nothing to act on. Returns a // failed result to short-circuit, or null when the target is present. private CommandResult? RequireTarget(string action, string target) @@ -714,6 +814,12 @@ public sealed class CommandExecutor // to exit and returns its real exit code, so success here means the script // actually ran to a zero exit - not merely that it was launched. Fails closed // when no interactive user is signed in. + // Upper bound on how long a user-session script may block the command + // worker. Generous, because a staged maintenance script can legitimately run + // for minutes, but finite: before this existed the wait was Infinite and a + // hung script pinned the caller forever. + private static readonly TimeSpan UserScriptWaitTimeout = TimeSpan.FromMinutes(5); + private CommandResult RunScriptInUserSession( string name, ScriptDeliveryClaim script, string scriptPath) { @@ -723,7 +829,8 @@ public sealed class CommandExecutor var workingDirectory = Path.GetDirectoryName(scriptPath) ?? ""; var outcome = InteractiveProcessLauncher.LaunchInActiveSession( - "powershell.exe", arguments, workingDirectory, waitForExit: true); + "powershell.exe", arguments, workingDirectory, waitForExit: true, + waitTimeout: UserScriptWaitTimeout); if (!outcome.Launched) { logger.LogWarning( @@ -737,6 +844,23 @@ public sealed class CommandExecutor }; } + if (outcome.TimedOut) + { + // The script is still running; we simply stopped waiting. Report it + // as unresolved rather than success, because we never saw an exit + // code and cannot claim it worked. + logger.LogWarning( + "run_script for {Name} still running after {Timeout}; stopped waiting", + name, UserScriptWaitTimeout); + return new CommandResult + { + Succeeded = false, + Message = $"Script {name} did not finish within " + + $"{UserScriptWaitTimeout.TotalMinutes:0} minutes and is still running", + Detail = ScriptDetail(name, script, scriptPath, exitCode: null), + }; + } + var exit = outcome.ExitCode; var ok = exit is null || exit == 0; return new CommandResult diff --git a/agent/ShopDbEndpointAgent.Core/Contracts.cs b/agent/ShopDbEndpointAgent.Core/Contracts.cs index d1c18f0f..bd2e448e 100644 --- a/agent/ShopDbEndpointAgent.Core/Contracts.cs +++ b/agent/ShopDbEndpointAgent.Core/Contracts.cs @@ -44,11 +44,29 @@ public sealed class EndpointReport // collection: both Uninstall registry hives, {displayname, displayversion, // publisher}. Server logs it as an unpersisted key until the SBOM store lands. [JsonPropertyName("applications")] public object[]? Applications { get; set; } + + // True when Windows is holding a reboot. A machine that has been waiting to + // reboot for weeks explains stalled patching and "the fix did not take" + // reports, and on a shop floor PC nobody reboots voluntarily. + [JsonPropertyName("pendingreboot")] public bool? PendingReboot { get; set; } + + // Session-scoped data gathered by a probe running as the signed-in user: + // mapped drives and user-owned listeners. Null when nobody is signed in or + // the probe did not produce a result this cycle - absent means "not + // observed", which is deliberately distinct from an empty list. + [JsonPropertyName("usercontext")] public object? UserContext { get; set; } } public sealed class EndpointEventRecord { [JsonPropertyName("eventid")] public int EventId { get; set; } + + // The Windows event record id: monotonic and unique within its channel on a + // given machine. It is both the agent's harvest watermark and the server's + // dedupe key, so the same record can never be stored twice. Null only when + // the source record did not expose one. + [JsonPropertyName("eventrecordid")] public long? EventRecordId { get; set; } + [JsonPropertyName("eventchannel")] public string? EventChannel { get; set; } [JsonPropertyName("eventsource")] public string? EventSource { get; set; } [JsonPropertyName("eventtime")] public DateTimeOffset? EventTime { get; set; } diff --git a/agent/ShopDbEndpointAgent.Core/CuratedEventCatalog.cs b/agent/ShopDbEndpointAgent.Core/CuratedEventCatalog.cs index 57b44d7f..143ce78d 100644 --- a/agent/ShopDbEndpointAgent.Core/CuratedEventCatalog.cs +++ b/agent/ShopDbEndpointAgent.Core/CuratedEventCatalog.cs @@ -27,7 +27,15 @@ public static class CuratedEventCatalog new(55, "System", "NTFS file system corruption"), new(7011, "System", "Service Control Manager timeout"), new(10016, "System", "DCOM permission error"), + // Windows Error Reporting on the Application channel: 1000 is the + // faulting-application record (app, version, faulting module, exception + // code), 1001 is the WER bucket record Windows pairs with it, and 1002 + // is an unresponsive app rather than a crash. The structured detail for + // all three arrives in the record's EventData, which the collector lifts + // into `fields`. + new(1000, "Application", "Application error / faulting module"), new(1001, "Application", "Application error / bugcheck report"), + new(1002, "Application", "Application hang"), }; // The distinct channels the collector must read. diff --git a/agent/ShopDbEndpointAgent.Core/EndpointCollector.cs b/agent/ShopDbEndpointAgent.Core/EndpointCollector.cs index 2486a66e..79e6a5ec 100644 --- a/agent/ShopDbEndpointAgent.Core/EndpointCollector.cs +++ b/agent/ShopDbEndpointAgent.Core/EndpointCollector.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using System.Diagnostics.Eventing.Reader; using System.Management; using System.Text.Json; using Microsoft.Extensions.Logging; @@ -19,16 +20,30 @@ public sealed class EndpointCollector : ICollector // keep the payload bounded on a chatty Security log. private const int MaxEventsPerChannel = 25; + // The user-context probe spawns a process in the operator's session, so it + // runs on its own slower sub-cadence rather than every telemetry tick. + // Mapped drives change at logon, not minute to minute, and this keeps the + // visible-session cost low on a shop floor PC. + private static readonly TimeSpan UserContextInterval = TimeSpan.FromMinutes(60); + private readonly ILogger logger; private readonly AgentConfig config; private readonly IAgentUtilityRunner? utilities; + private readonly EventWatermarkStore? watermarks; + private readonly UserContextProbe? userContextProbe; + + private DateTimeOffset lastUserContextAttempt = DateTimeOffset.MinValue; + private object? lastUserContext; public EndpointCollector(ILogger logger, AgentConfig config, - IAgentUtilityRunner? utilities = null) + IAgentUtilityRunner? utilities = null, EventWatermarkStore? watermarks = null, + UserContextProbe? userContextProbe = null) { this.logger = logger; this.config = config; this.utilities = utilities; + this.watermarks = watermarks; + this.userContextProbe = userContextProbe; } public EndpointReport Collect(string agentVersion, bool adopt) @@ -49,6 +64,7 @@ public sealed class EndpointCollector : ICollector MachineNumber = GetMachineNumber(), LastBootTime = GetLastBootTime(), Services = GetMonitoredTargets(), + Ports = GetListeningPorts(), Interfaces = GetInterfaces(), Events = GetCuratedEvents(), Disks = GetDisks(), @@ -58,14 +74,206 @@ public sealed class EndpointCollector : ICollector OsInfo = GetOsInfo(), Domain = GetDomain(), Applications = GetApplications(), + PendingReboot = GetPendingReboot(), + UserContext = GetUserContext(adopt), }; if (report.LastBootTime is DateTimeOffset boot) report.UptimeSeconds = (long)(DateTimeOffset.UtcNow - boot).TotalSeconds; + report.CpuPercent = GetCpuPercent(); + report.MemoryPercent = GetMemoryPercent(); + report.IdleSeconds = GetIdleSecondsFromUserContext(); + return report; } + // Instantaneous CPU load from the _Total performance instance. + // + // Uses the WMI formatted-data class rather than PerformanceCounter because a + // PerformanceCounter reads 0 on its first sample and needs a second sample a + // second later to mean anything. Sleeping a second inside the collect path + // would delay the heartbeat on every tick for one number. The formatted-data + // class is already averaged by the OS, so a single read is usable. + private double? GetCpuPercent() + { + try + { + using var searcher = new ManagementObjectSearcher( + "SELECT PercentProcessorTime FROM " + + "Win32_PerfFormattedData_PerfOS_Processor WHERE Name='_Total'"); + foreach (var mo in searcher.Get()) + { + var value = ToLong(mo["PercentProcessorTime"]); + if (value is long percent) + return Math.Clamp(percent, 0, 100); + } + } + catch (Exception error) + { + logger.LogDebug(error, "CPU percent query failed"); + } + return null; + } + + // Percentage of physical memory in use. Reported alongside the absolute + // byte counts in the memory object because the percentage is what an alert + // threshold keys off, and deriving it server-side would break whenever the + // byte fields are absent. + private double? GetMemoryPercent() + { + try + { + using var searcher = new ManagementObjectSearcher( + "SELECT TotalVisibleMemorySize, FreePhysicalMemory " + + "FROM Win32_OperatingSystem"); + foreach (var mo in searcher.Get()) + { + var totalKb = ToLong(mo["TotalVisibleMemorySize"]); + var freeKb = ToLong(mo["FreePhysicalMemory"]); + if (totalKb is long total && total > 0 && freeKb is long free) + { + var used = (double)(total - free) / total * 100.0; + return Math.Round(Math.Clamp(used, 0, 100), 1); + } + } + } + catch (Exception error) + { + logger.LogDebug(error, "Memory percent query failed"); + } + return null; + } + + // Idle time is measured by the user-context probe, because GetLastInputInfo + // is per session and this service runs in session 0. Null when no probe + // result is available, which correctly reads as "not observed" rather than + // "the machine has been idle for zero seconds". + private long? GetIdleSecondsFromUserContext() + { + if (lastUserContext is not Dictionary context) + return null; + if (!context.TryGetValue("idleseconds", out var raw) || raw is null) + return null; + + try + { + if (raw is JsonElement element) + { + return element.ValueKind == JsonValueKind.Number + && element.TryGetInt64(out var parsed) ? parsed : null; + } + return Convert.ToInt64(raw); + } + catch (Exception error) + { + logger.LogDebug(error, "Could not read idleseconds from the user context"); + return null; + } + } + + // Listening TCP ports with the owning process name. + // + // Only listeners, never established connections: listeners answer "what + // service is this machine offering" (the DNC port, the MTConnect adapter), + // which is stable and small. Established connections churn constantly and + // would make every report differ from the last for no diagnostic gain, so + // the full table is exposed as an on-demand command instead. + private const int MaxListeners = 100; + + private object[] GetListeningPorts() + { + try + { + var listeners = TcpTableReader.GetListeners(); + if (listeners.Count == 0) + return Array.Empty(); + + var names = GetProcessNames(); + var result = new List(); + + foreach (var listener in listeners) + { + if (result.Count >= MaxListeners) + { + logger.LogDebug( + "Listener list truncated at {Limit} entries.", MaxListeners); + break; + } + + names.TryGetValue(listener.ProcessId, out var processName); + result.Add(new + { + localaddress = listener.LocalAddress, + localport = listener.LocalPort, + processid = listener.ProcessId, + processname = processName, + protocol = "tcp", + }); + } + + return result.ToArray(); + } + catch (Exception error) + { + logger.LogDebug(error, "Listening port enumeration failed"); + return Array.Empty(); + } + } + + // Snapshot the process table once and index by id. Resolving each listener + // individually would be an O(n) call per row and can race with process exit. + private Dictionary GetProcessNames() + { + var names = new Dictionary(); + try + { + foreach (var process in Process.GetProcesses()) + { + try + { + names[process.Id] = process.ProcessName; + } + catch + { + // The process exited between enumeration and read. + } + finally + { + process.Dispose(); + } + } + } + catch (Exception error) + { + logger.LogDebug(error, "Process name snapshot failed"); + } + return names; + } + + // Session-scoped data on a slow sub-cadence. Adoption forces a fresh probe so + // a newly enrolled machine reports its mapped drives immediately instead of + // looking bare for the first hour. Between probes the previous result is + // repeated, because a mapped drive stays mapped and dropping the key would + // read on the server as "the drive went away". + private object? GetUserContext(bool force) + { + if (userContextProbe is null) + return null; + + var due = force + || DateTimeOffset.UtcNow - lastUserContextAttempt >= UserContextInterval; + if (!due) + return lastUserContext; + + lastUserContextAttempt = DateTimeOffset.UtcNow; + var collected = userContextProbe.TryCollect(); + if (collected is not null) + lastUserContext = collected; + + return lastUserContext; + } + // The interactive tray app ships and updates independently of this service, // so the agent reports the tray's file version as its own field. Resolve the // tray exe beside the agent first (the normal install layout), then the @@ -451,40 +659,33 @@ public sealed class EndpointCollector : ICollector } } - // Walk each curated channel newest-first, keep only the curated IDs for - // that channel, cap per channel. Reads Security/System/Application - the - // old scaffold read System only and filtered by the wrong IDs. + // Walk each curated channel newest-first and report ONLY records newer than + // the stored watermark, so a record crosses the wire exactly once. + // + // Two things changed here versus the original scaffold. First, filtering by + // event id now happens inside the event log service through an XPath query + // instead of walking every entry in the channel from C#; on a busy Security + // log that is the difference between touching a handful of records and + // hundreds of thousands. Second, EventLogReader (not the legacy EventLog + // API) exposes RecordId, which is what makes exactly-once possible. + // + // First contact on a channel has no watermark, so it sends a bounded tail + // (MaxEventsPerChannel) for initial context rather than the whole backlog. + // The same cap also bounds a burst: if more than the cap arrived since the + // last cycle, the NEWEST are taken and the watermark still jumps to the + // newest, so the payload can never balloon. private EndpointEventRecord[] GetCuratedEvents() { var records = new List(); foreach (var channel in CuratedEventCatalog.Channels) { var wanted = CuratedEventCatalog.IdsForChannel(channel); + if (wanted.Count == 0) + continue; + try { - using var log = new EventLog(channel); - int kept = 0; - for (int i = log.Entries.Count - 1; - i >= 0 && kept < MaxEventsPerChannel; i--) - { - EventLogEntry entry; - try { entry = log.Entries[i]; } catch { continue; } - - int id = (int)(entry.InstanceId & 0xFFFF); - if (!wanted.Contains(id)) - continue; - - records.Add(new EndpointEventRecord - { - EventId = id, - EventChannel = channel, - EventSource = entry.Source, - EventTime = new DateTimeOffset( - entry.TimeGenerated.ToUniversalTime(), TimeSpan.Zero), - EventMessage = Truncate(entry.Message, 1000), - }); - kept++; - } + records.AddRange(ReadChannel(channel, wanted)); } catch (Exception error) { @@ -493,9 +694,172 @@ public sealed class EndpointCollector : ICollector logger.LogDebug(error, "Event channel {Channel} unreadable", channel); } } + + watermarks?.Save(); return records.ToArray(); } + private IEnumerable ReadChannel( + string channel, IReadOnlySet wanted) + { + long watermark = watermarks?.GetWatermark(channel) ?? 0; + bool seeding = watermarks is null || !watermarks.HasWatermark(channel); + + var query = new EventLogQuery(channel, PathType.LogName, BuildIdFilter(wanted)) + { + // Newest-first so the cap keeps the most recent records and we can + // stop as soon as we reach the watermark. + ReverseDirection = true, + }; + + var found = new List(); + long newestSeen = 0; + + using var reader = new EventLogReader(query); + while (found.Count < MaxEventsPerChannel) + { + EventRecord? entry; + try { entry = reader.ReadEvent(); } + catch (Exception error) + { + logger.LogDebug(error, "Event read failed on {Channel}", channel); + break; + } + + if (entry is null) + break; + + using (entry) + { + long recordId = entry.RecordId ?? 0; + if (recordId > newestSeen) + newestSeen = recordId; + + // Reverse order means everything from here down is older than + // the watermark and has already been reported. + if (!seeding && recordId != 0 && recordId <= watermark) + break; + + found.Add(ToEventRecord(entry, channel, recordId)); + } + } + + // A cleared log restarts record ids low; reseed rather than go silent. + if (!seeding && newestSeen > 0 + && watermarks is not null && watermarks.ResetIfStale(channel, newestSeen)) + { + return found; + } + + if (newestSeen > 0) + watermarks?.Advance(channel, newestSeen); + + return found; + } + + private EndpointEventRecord ToEventRecord( + EventRecord entry, string channel, long recordId) + { + string? message = null; + try { message = entry.FormatDescription(); } + catch (Exception error) + { + // A provider whose message DLL is missing cannot render text; the + // id, time, and source are still worth reporting. + logger.LogDebug(error, "Event message not renderable on {Channel}", channel); + } + + return new EndpointEventRecord + { + EventId = entry.Id, + EventRecordId = recordId == 0 ? null : recordId, + EventChannel = channel, + EventSource = entry.ProviderName, + EventTime = entry.TimeCreated is DateTime created + ? new DateTimeOffset(created.ToUniversalTime(), TimeSpan.Zero) + : null, + EventMessage = Truncate(message, 1000), + Fields = GetEventFields(entry, channel), + }; + } + + // Bounds on the structured field extraction. The rendered message is often + // a paragraph of prose while the useful specifics (which account, which + // logon type, which faulting module) live in EventData, so it is worth + // carrying - but a single record can declare a lot of elements and some + // providers emit large blobs, so both the count and each value are capped. + private const int MaxEventFields = 30; + private const int MaxEventFieldLength = 512; + + // Pull the provider's structured payload out of the event XML into the + // existing `fields` dictionary, which the server already stores as + // payloadjson. Most providers use v; + // some use with an arbitrary provider-defined subtree, so both + // are walked. Unnamed elements keep their positional index as the + // key, because for several catalog IDs position IS the meaning. + // + // Never throws: a provider with malformed or unreadable XML degrades to no + // fields rather than losing the whole record. + private Dictionary? GetEventFields(EventRecord entry, string channel) + { + try + { + var xml = entry.ToXml(); + if (string.IsNullOrWhiteSpace(xml)) + return null; + + var document = System.Xml.Linq.XDocument.Parse(xml); + var root = document.Root; + if (root is null) + return null; + + var container = + root.Elements().FirstOrDefault(e => e.Name.LocalName == "EventData") + ?? root.Elements().FirstOrDefault(e => e.Name.LocalName == "UserData"); + if (container is null) + return null; + + var fields = new Dictionary(); + int unnamed = 0; + foreach (var element in container.Descendants()) + { + if (fields.Count >= MaxEventFields) + break; + if (element.HasElements) + continue; + + var value = element.Value; + if (string.IsNullOrWhiteSpace(value)) + continue; + + var name = element.Attribute("Name")?.Value; + if (string.IsNullOrWhiteSpace(name)) + { + name = element.Name.LocalName == "Data" + ? "data" + unnamed++ + : element.Name.LocalName; + } + + fields[name!] = Truncate(value, MaxEventFieldLength) ?? string.Empty; + } + + return fields.Count > 0 ? fields : null; + } + catch (Exception error) + { + logger.LogDebug(error, "Event XML not parseable on {Channel}", channel); + return null; + } + } + + // An XPath the event log service evaluates itself, e.g. + // *[System[(EventID=4624 or EventID=4625)]] + private static string BuildIdFilter(IReadOnlySet wanted) + { + var clauses = string.Join(" or ", wanted.Select(id => "EventID=" + id)); + return "*[System[(" + clauses + ")]]"; + } + // Installed-software SBOM. Mirrors the Assess-Orchestrator applications // collection exactly: enumerate both Uninstall registry hives (64-bit and // WOW6432Node), keep only entries with a DisplayName, and emit @@ -566,6 +930,51 @@ public sealed class EndpointCollector : ICollector } } + // Whether Windows is waiting on a reboot. + // + // There is no single API for this, so we check the three signals Windows + // itself uses: the CBS servicing flag, the Windows Update auto-update flag, + // and a queued file-rename operation. Any one of them is sufficient. Checked + // by key EXISTENCE rather than value, which is how these flags work - the + // key is created when a reboot is needed and deleted once it happens. + private bool? GetPendingReboot() + { + try + { + const string cbs = + @"SOFTWARE\Microsoft\Windows\CurrentVersion\Component Based Servicing\RebootPending"; + const string windowsUpdate = + @"SOFTWARE\Microsoft\Windows\CurrentVersion\WindowsUpdate\Auto Update\RebootRequired"; + + if (LocalMachineKeyExists(cbs) || LocalMachineKeyExists(windowsUpdate)) + return true; + + using var sessionManager = Registry.LocalMachine.OpenSubKey( + @"SYSTEM\CurrentControlSet\Control\Session Manager"); + if (sessionManager?.GetValue("PendingFileRenameOperations") is string[] pending + && pending.Length > 0) + { + return true; + } + + return false; + } + catch (Exception error) + { + // Null, not false: an unreadable registry means we do not know, and + // reporting a confident "no reboot pending" would be worse than + // reporting nothing. + logger.LogDebug(error, "Pending reboot check failed"); + return null; + } + } + + private static bool LocalMachineKeyExists(string path) + { + using var key = Registry.LocalMachine.OpenSubKey(path); + return key is not null; + } + // Fixed local disks with capacity/free so the server can flag a PC that is // running out of space. Win32_LogicalDisk DriveType 3 = local fixed disk; // network/removable drives are skipped. Bytes stay as longs; the server can @@ -591,6 +1000,14 @@ public sealed class EndpointCollector : ICollector freebytes = free, usedbytes = size is long total && free is long available ? total - available : (long?)null, + // Precomputed so an alert threshold is a direct comparison. + // A 40GB free reading means nothing without the capacity it + // came from, and the server would have to guard against a + // null or zero size on every evaluation to derive it. + freepercent = size is long capacity && capacity > 0 + && free is long remaining + ? Math.Round((double)remaining / capacity * 100.0, 1) + : (double?)null, }); } } diff --git a/agent/ShopDbEndpointAgent.Core/EventWatermarkStore.cs b/agent/ShopDbEndpointAgent.Core/EventWatermarkStore.cs new file mode 100644 index 00000000..ba4cfdd4 --- /dev/null +++ b/agent/ShopDbEndpointAgent.Core/EventWatermarkStore.cs @@ -0,0 +1,132 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging; + +namespace ShopDbEndpointAgent.Core; + +// Remembers the newest Windows event record id already reported for each +// curated channel, so a collect cycle harvests ONLY records the server has not +// seen. +// +// Why this exists: the collector used to re-read the newest N records of every +// channel on every cycle, and the server ingest appends event rows without a +// dedupe check. A single crash therefore landed in the database once per +// cadence tick for as long as it stayed inside that window - up to 96 identical +// rows a day per event. Carrying a watermark makes the steady state "only what +// is new", which is also what makes the full event XML affordable to send. +// +// The watermark is persisted under %ProgramData% (not in memory) because a +// service restart would otherwise re-report the whole window again, which is +// exactly the crash-loop case where duplicate events are most likely. +// +// Record ids are per channel and monotonic on a given machine. A cleared or +// rebuilt event log restarts them at 1; that is handled by treating a stored +// watermark HIGHER than the newest available record as stale and resetting it, +// otherwise the channel would go permanently silent. +public sealed class EventWatermarkStore +{ + private const string StoreFileName = "eventwatermark.json"; + + // %ProgramData%\ShopDB\EndpointAgent - the same root the rolling agent log + // lives under, so all agent-owned state sits in one place. + public static string DefaultStateDirectory => Path.Combine( + Environment.GetFolderPath(Environment.SpecialFolder.CommonApplicationData), + "ShopDB", "EndpointAgent"); + + private readonly string storePath; + private readonly ILogger logger; + private readonly Dictionary watermarks; + + public EventWatermarkStore(ILogger logger, + string? stateDirectory = null) + { + this.logger = logger; + var directory = string.IsNullOrWhiteSpace(stateDirectory) + ? DefaultStateDirectory + : stateDirectory; + storePath = Path.Combine(directory, StoreFileName); + watermarks = Load(); + } + + // The last record id reported for this channel, or 0 when the channel has + // never been harvested. 0 means "first contact": the caller sends a bounded + // tail for initial context rather than the entire backlog. + public long GetWatermark(string channel) + => watermarks.TryGetValue(channel, out var value) ? value : 0; + + public bool HasWatermark(string channel) + => watermarks.ContainsKey(channel); + + // Advance the channel forward only. Never moves backwards, so an + // out-of-order read cannot cause the same record to be reported twice. + public void Advance(string channel, long recordId) + { + if (recordId <= 0) + return; + if (watermarks.TryGetValue(channel, out var current) && current >= recordId) + return; + watermarks[channel] = recordId; + } + + // A cleared or rebuilt log restarts record ids near 1. When the newest + // record available is below the stored watermark the store is stale, so + // drop it and let the channel re-seed instead of going silent forever. + public bool ResetIfStale(string channel, long newestAvailableRecordId) + { + if (newestAvailableRecordId <= 0) + return false; + if (!watermarks.TryGetValue(channel, out var current) || current <= newestAvailableRecordId) + return false; + + logger.LogInformation( + "Event channel {Channel} record id went backwards ({Stored} -> {Newest}); log was cleared, reseeding", + channel, current, newestAvailableRecordId); + watermarks.Remove(channel); + return true; + } + + // Best-effort persist. A failure here costs at most a repeated report of + // records already sent, so it must never abort a collection. + public void Save() + { + try + { + var directory = Path.GetDirectoryName(storePath); + if (!string.IsNullOrWhiteSpace(directory)) + Directory.CreateDirectory(directory); + + // Write-then-rename so a crash mid-write cannot leave a truncated + // file that would reseed every channel on the next start. + var temporaryPath = storePath + ".tmp"; + File.WriteAllText(temporaryPath, + JsonSerializer.Serialize(watermarks)); + File.Move(temporaryPath, storePath, overwrite: true); + } + catch (Exception error) + { + logger.LogDebug(error, "Could not persist event watermarks to {Path}", storePath); + } + } + + private Dictionary Load() + { + try + { + if (!File.Exists(storePath)) + return new Dictionary(StringComparer.OrdinalIgnoreCase); + + var loaded = JsonSerializer.Deserialize>( + File.ReadAllText(storePath)); + return loaded is null + ? new Dictionary(StringComparer.OrdinalIgnoreCase) + : new Dictionary(loaded, StringComparer.OrdinalIgnoreCase); + } + catch (Exception error) + { + // A corrupt store reseeds every channel, which costs one bounded + // tail per channel - acceptable versus failing the collection. + logger.LogWarning(error, + "Event watermark store unreadable at {Path}; reseeding channels", storePath); + return new Dictionary(StringComparer.OrdinalIgnoreCase); + } + } +} diff --git a/agent/ShopDbEndpointAgent.Core/InteractiveProcessLauncher.cs b/agent/ShopDbEndpointAgent.Core/InteractiveProcessLauncher.cs index 326bdc07..b9c45755 100644 --- a/agent/ShopDbEndpointAgent.Core/InteractiveProcessLauncher.cs +++ b/agent/ShopDbEndpointAgent.Core/InteractiveProcessLauncher.cs @@ -1,5 +1,6 @@ using System.ComponentModel; using System.Runtime.InteropServices; +using System.Text; namespace ShopDbEndpointAgent.Core; @@ -18,7 +19,11 @@ namespace ShopDbEndpointAgent.Core; // Only process_start uses this. Stop/status/service actions stay as SYSTEM. public static class InteractiveProcessLauncher { - public readonly record struct LaunchOutcome(bool Launched, string? Error, int? ExitCode = null); + // TimedOut=true means the process launched fine but was still running when + // the caller's wait budget expired. It is deliberately NOT killed, so the + // work still completes; the caller just stops blocking on it. + public readonly record struct LaunchOutcome( + bool Launched, string? Error, int? ExitCode = null, bool TimedOut = false); private const int TokenDuplicate = 0x0002; private const int TokenQuery = 0x0008; @@ -34,6 +39,10 @@ public static class InteractiveProcessLauncher private const uint Infinite = 0xFFFFFFFF; + // WaitForSingleObject returns this when the budget expires with the handle + // still unsignalled. + private const uint WaitTimeout = 0x00000102; + // Returns Launched=true when CreateProcessAsUser succeeds. Never throws; // failures come back as Launched=false with a descriptive Error so the // caller can surface a clean CommandResult. @@ -51,9 +60,17 @@ public static class InteractiveProcessLauncher // needs the actual outcome (run_script in the user session) can report // completed/failed instead of fire-and-forget "launched". GUI callers // (process_start) leave it false and return as soon as the window is up. + // + // waitTimeout bounds that wait. This matters because the caller may be the + // telemetry loop, and telemetry IS the heartbeat: a script that hangs in the + // user session used to block here forever (the wait was hardcoded to + // Infinite), which would stall reporting and make a perfectly healthy PC + // read as offline in the WebUI. On expiry the child is left running and the + // outcome comes back with TimedOut=true. Null means wait forever and should + // only be used by a caller that is not on a timed path. public static LaunchOutcome LaunchInActiveSession( string applicationPath, string? arguments, string? workingDirectory, - bool waitForExit = false) + bool waitForExit = false, TimeSpan? waitTimeout = null) { var sessionId = WTSGetActiveConsoleSessionId(); if (sessionId == InvalidSessionId) @@ -129,11 +146,25 @@ public static class InteractiveProcessLauncher } int? exitCode = null; + bool timedOut = false; if (waitForExit && processInformation.hProcess != IntPtr.Zero) { - WaitForSingleObject(processInformation.hProcess, Infinite); - if (GetExitCodeProcess(processInformation.hProcess, out var code)) + uint milliseconds = waitTimeout is TimeSpan budget + ? (uint)Math.Clamp(budget.TotalMilliseconds, 0, uint.MaxValue - 1) + : Infinite; + + if (WaitForSingleObject(processInformation.hProcess, milliseconds) + == WaitTimeout) + { + // Still running. Leave it alone - killing it would abandon + // half-done work, and a file-spooling probe will still write + // its result for the next cycle to pick up. + timedOut = true; + } + else if (GetExitCodeProcess(processInformation.hProcess, out var code)) + { exitCode = unchecked((int)code); + } } if (processInformation.hProcess != IntPtr.Zero) @@ -141,7 +172,7 @@ public static class InteractiveProcessLauncher if (processInformation.hThread != IntPtr.Zero) CloseHandle(processInformation.hThread); - return new LaunchOutcome(true, null, exitCode); + return new LaunchOutcome(true, null, exitCode, timedOut); } finally { @@ -154,6 +185,57 @@ public static class InteractiveProcessLauncher } } + // Profile directory of the active console user, for example + // C:\Users\ShopFloor. The agent runs as LocalSystem, so expanding + // %USERPROFILE% here would return SYSTEM's own profile under + // config\systemprofile rather than the operator's. The interactive user's + // token is the only reliable source, and this class already knows how to + // obtain it. Returns null with a reason when nobody is signed in. + public static string? TryGetActiveUserProfileDirectory(out string? error) + { + var sessionId = WTSGetActiveConsoleSessionId(); + if (sessionId == InvalidSessionId) + { + error = "No active console session (no interactive user is signed in)."; + return null; + } + + if (!WTSQueryUserToken(sessionId, out var userToken)) + { + error = "Could not obtain the interactive user token: " + + new Win32Exception(Marshal.GetLastWin32Error()).Message; + return null; + } + + try + { + uint size = 0; + GetUserProfileDirectory(userToken, null, ref size); + if (size == 0) + { + error = "Could not size the user profile directory: " + + new Win32Exception(Marshal.GetLastWin32Error()).Message; + return null; + } + + var buffer = new StringBuilder((int)size); + if (!GetUserProfileDirectory(userToken, buffer, ref size)) + { + error = "Could not resolve the user profile directory: " + + new Win32Exception(Marshal.GetLastWin32Error()).Message; + return null; + } + + error = null; + return buffer.ToString(); + } + finally + { + if (userToken != IntPtr.Zero) + CloseHandle(userToken); + } + } + private enum SecurityImpersonationLevel { SecurityAnonymous, @@ -222,6 +304,10 @@ public static class InteractiveProcessLauncher [DllImport("userenv.dll", SetLastError = true)] private static extern bool DestroyEnvironmentBlock(IntPtr environment); + [DllImport("userenv.dll", SetLastError = true, CharSet = CharSet.Unicode)] + private static extern bool GetUserProfileDirectory( + IntPtr token, StringBuilder? profileDirectory, ref uint size); + [DllImport("advapi32.dll", SetLastError = true, CharSet = CharSet.Unicode)] private static extern bool CreateProcessAsUser( IntPtr token, diff --git a/agent/ShopDbEndpointAgent.Core/TcpTableReader.cs b/agent/ShopDbEndpointAgent.Core/TcpTableReader.cs new file mode 100644 index 00000000..26af6257 --- /dev/null +++ b/agent/ShopDbEndpointAgent.Core/TcpTableReader.cs @@ -0,0 +1,128 @@ +using System.Net; +using System.Runtime.InteropServices; + +namespace ShopDbEndpointAgent.Core; + +// TCP connection state with owning process ids, via GetExtendedTcpTable. +// +// Why P/Invoke instead of shelling out to netstat.exe: parsing netstat text +// output means spawning a process on every telemetry tick and depending on the +// column layout of a localized command. GetExtendedTcpTable is the API netstat +// itself calls, returns process ids directly, and costs a single allocation. +// +// Listeners are the cheap, bounded half of this (a handful of rows, stable over +// time) and ride the normal cadence. The full connection table is unbounded and +// churns constantly, so it stays on-demand. +public static class TcpTableReader +{ + private const int AF_INET = 2; + private const uint ERROR_INSUFFICIENT_BUFFER = 122; + private const int TCP_TABLE_OWNER_PID_ALL = 5; + + public enum TcpState + { + Closed = 1, + Listen = 2, + SynSent = 3, + SynReceived = 4, + Established = 5, + FinWait1 = 6, + FinWait2 = 7, + CloseWait = 8, + Closing = 9, + LastAck = 10, + TimeWait = 11, + DeleteTcb = 12, + } + + public readonly record struct TcpRow( + string LocalAddress, + int LocalPort, + string RemoteAddress, + int RemotePort, + TcpState State, + int ProcessId); + + // Full IPv4 TCP table. Throws only on catastrophic interop failure; callers + // treat an empty list as "not observed". + public static List GetTcpTable() + { + var rows = new List(); + var buffer = IntPtr.Zero; + var size = 0; + + try + { + // First call sizes the buffer. It is expected to fail with + // INSUFFICIENT_BUFFER; anything else is a real error. + var result = GetExtendedTcpTable( + IntPtr.Zero, ref size, false, AF_INET, TCP_TABLE_OWNER_PID_ALL, 0); + if (result != ERROR_INSUFFICIENT_BUFFER && result != 0) + return rows; + + buffer = Marshal.AllocHGlobal(size); + result = GetExtendedTcpTable( + buffer, ref size, false, AF_INET, TCP_TABLE_OWNER_PID_ALL, 0); + if (result != 0) + return rows; + + var entryCount = Marshal.ReadInt32(buffer); + var rowSize = Marshal.SizeOf(); + var cursor = IntPtr.Add(buffer, 4); + + for (var index = 0; index < entryCount; index++) + { + var row = Marshal.PtrToStructure(cursor); + rows.Add(new TcpRow( + FormatAddress(row.localAddress), + FormatPort(row.localPort), + FormatAddress(row.remoteAddress), + FormatPort(row.remotePort), + (TcpState)row.state, + (int)row.owningProcessId)); + cursor = IntPtr.Add(cursor, rowSize); + } + } + finally + { + if (buffer != IntPtr.Zero) + Marshal.FreeHGlobal(buffer); + } + + return rows; + } + + public static List GetListeners() + => GetTcpTable().FindAll(row => row.State == TcpState.Listen); + + private static string FormatAddress(uint address) + => new IPAddress(BitConverter.GetBytes(address)).ToString(); + + // The port arrives as a big-endian value packed into the low two bytes of a + // host-order uint, so the bytes must be swapped rather than simply cast. + private static int FormatPort(uint port) + { + var bytes = BitConverter.GetBytes(port); + return (bytes[0] << 8) | bytes[1]; + } + + [StructLayout(LayoutKind.Sequential)] + private struct MIB_TCPROW_OWNER_PID + { + public uint state; + public uint localAddress; + public uint localPort; + public uint remoteAddress; + public uint remotePort; + public uint owningProcessId; + } + + [DllImport("iphlpapi.dll", SetLastError = true)] + private static extern uint GetExtendedTcpTable( + IntPtr tcpTable, + ref int size, + bool order, + int addressFamily, + int tableClass, + int reserved); +} diff --git a/agent/ShopDbEndpointAgent.Core/UserContextProbe.cs b/agent/ShopDbEndpointAgent.Core/UserContextProbe.cs new file mode 100644 index 00000000..c25d178c --- /dev/null +++ b/agent/ShopDbEndpointAgent.Core/UserContextProbe.cs @@ -0,0 +1,202 @@ +using Microsoft.Extensions.Logging; + +namespace ShopDbEndpointAgent.Core; + +// Runs a short PowerShell probe as the interactive user and returns what it +// wrote to the spool. +// +// Collects only data that genuinely does not exist in the agent's own context: +// netuse - mapped drives are per logon session; "net use" from +// LocalSystem returns SYSTEM's (empty) mapping table, never the +// operator's Z: to the file server. +// userlisteners- TCP listeners owned by the signed-in user's processes, which +// is what identifies an operator-launched DNC or CMM listener. +// idleseconds - GetLastInputInfo is per session, so session 0 cannot answer +// it. Sampled at probe time only, so treat it as a coarse +// "was anyone at the machine" signal, not live presence. +// +// The script is embedded rather than shipped as a .ps1 so the probe cannot drift +// from the agent binary or be tampered with on disk between cycles. +public sealed class UserContextProbe +{ + // The probe is bounded well below the telemetry cadence. If it cannot finish + // in this time the cycle simply reports no session data; it never delays the + // heartbeat. + private static readonly TimeSpan ProbeTimeout = TimeSpan.FromSeconds(45); + + private const string ProbeScript = """ + $ErrorActionPreference = 'Stop' + $result = [ordered]@{} + $spool = Join-Path $env:USERPROFILE 'ShopDB' + $target = Join-Path $spool 'usercontext.json' + $temporary = "$target.tmp" + + try { + New-Item -ItemType Directory -Path $spool -Force | Out-Null + + $netuse = @() + try { + Get-CimInstance -ClassName Win32_NetworkConnection -ErrorAction Stop | + ForEach-Object { + $netuse += [ordered]@{ + localname = $_.LocalName + remotename = $_.RemoteName + status = $_.Status + username = $_.UserName + persistent = [bool]$_.Persistent + } + } + } catch { + $result['netuseerror'] = $_.Exception.Message + } + $result['netuse'] = $netuse + + $listeners = @() + try { + $ownprocesses = @{} + Get-Process -ErrorAction SilentlyContinue | ForEach-Object { + $ownprocesses[$_.Id] = $_.ProcessName + } + Get-NetTCPConnection -State Listen -ErrorAction Stop | + Where-Object { $ownprocesses.ContainsKey($_.OwningProcess) } | + ForEach-Object { + $listeners += [ordered]@{ + localaddress = $_.LocalAddress + localport = $_.LocalPort + processid = $_.OwningProcess + processname = $ownprocesses[$_.OwningProcess] + } + } + } catch { + $result['userlistenerserror'] = $_.Exception.Message + } + $result['userlisteners'] = $listeners + + # Idle time must be measured inside the user's session: + # GetLastInputInfo is per session, so asking from the agent's + # session 0 would report the service's own idle time, which is + # meaningless. This is what distinguishes "powered on" from + # "someone is actually working at this machine". + try { + # Built as a single-quoted string rather than a here-string: a + # here-string terminator must sit at column 0, which cannot be + # expressed inside an indented C# raw string literal. + $signature = 'using System; using System.Runtime.InteropServices; ' + + 'public class ShopDbIdle { ' + + '[StructLayout(LayoutKind.Sequential)] ' + + 'struct LASTINPUTINFO { public uint cbSize; public uint dwTime; } ' + + '[DllImport("user32.dll")] ' + + 'static extern bool GetLastInputInfo(ref LASTINPUTINFO info); ' + + 'public static long IdleSeconds() { ' + + 'LASTINPUTINFO info = new LASTINPUTINFO(); ' + + 'info.cbSize = (uint)Marshal.SizeOf(info); ' + + 'if (!GetLastInputInfo(ref info)) return -1; ' + + 'return (long)((uint)Environment.TickCount - info.dwTime) / 1000; } }' + Add-Type -TypeDefinition $signature -ErrorAction Stop + $idle = [ShopDbIdle]::IdleSeconds() + if ($idle -ge 0) { $result['idleseconds'] = $idle } + } catch { + $result['idlesecondserror'] = $_.Exception.Message + } + + $result['collectedat'] = (Get-Date).ToUniversalTime().ToString('o') + $result['username'] = $env:USERNAME + + # Write to a temporary name and rename into place, so the reader can + # never observe a half-written file and discard a good result. + $result | ConvertTo-Json -Depth 4 -Compress | + Set-Content -Path $temporary -Encoding UTF8 -NoNewline + Move-Item -Path $temporary -Destination $target -Force + } catch { + exit 1 + } + exit 0 + """; + + private readonly UserContextSpool spool; + private readonly ILogger logger; + + public UserContextProbe(UserContextSpool spool, ILogger logger) + { + this.spool = spool; + this.logger = logger; + } + + // Returns the probe payload, or null when there is no interactive session or + // the probe produced nothing usable. Never throws: session-scoped data is a + // nice-to-have and must not be able to fail a telemetry cycle. + public Dictionary? TryCollect() + { + string? scriptPath = null; + try + { + var resultPath = spool.TryGetResultPath(out var reason); + if (resultPath is null) + { + logger.LogDebug( + "Skipping user-context collection: {Reason}", reason); + return null; + } + + spool.EnsureSpoolDirectory(resultPath); + spool.ClearStaleResult(resultPath); + + scriptPath = Path.Combine( + Path.GetTempPath(), $"shopdb-usercontext-{Guid.NewGuid():N}.ps1"); + File.WriteAllText(scriptPath, ProbeScript); + + var arguments = + "-NoProfile -NonInteractive -ExecutionPolicy Bypass -WindowStyle Hidden -File " + + "\"" + scriptPath + "\""; + + var outcome = InteractiveProcessLauncher.LaunchInActiveSession( + "powershell.exe", arguments, Path.GetTempPath(), waitForExit: true, + waitTimeout: ProbeTimeout); + + if (!outcome.Launched) + { + logger.LogDebug( + "The user-context probe did not launch: {Error}", outcome.Error); + return null; + } + + if (outcome.TimedOut) + { + logger.LogWarning( + "The user-context probe did not finish within {Seconds} seconds.", + ProbeTimeout.TotalSeconds); + return null; + } + + if (outcome.ExitCode is int exit && exit != 0) + { + logger.LogWarning( + "The user-context probe exited with code {ExitCode}.", exit); + return null; + } + + return spool.TryConsumeResult(resultPath); + } + catch (Exception exception) + { + logger.LogWarning(exception, "The user-context probe failed."); + return null; + } + finally + { + if (scriptPath is not null) + { + try + { + if (File.Exists(scriptPath)) + File.Delete(scriptPath); + } + catch + { + // A leftover temp script is harmless; it is overwritten by + // name on the next cycle and cleaned by the OS. + } + } + } + } +} diff --git a/agent/ShopDbEndpointAgent.Core/UserContextSpool.cs b/agent/ShopDbEndpointAgent.Core/UserContextSpool.cs new file mode 100644 index 00000000..36dff117 --- /dev/null +++ b/agent/ShopDbEndpointAgent.Core/UserContextSpool.cs @@ -0,0 +1,149 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging; + +namespace ShopDbEndpointAgent.Core; + +// Reads the result file that a user-context probe writes into the signed-in +// user's ShopDB spool folder, then removes it. +// +// Why a file and not a return value: the agent is a session-0 LocalSystem +// service, so anything session-scoped (mapped drives from "net use", per-user +// network state) simply does not exist in its own context. The probe has to run +// as the interactive user, and a process launched with CreateProcessAsUser gives +// back an exit code, not stdout. A file in the user's own profile is the +// handoff, and the user can always write there without elevation. +// +// Ordering matters here. The stale file is deleted BEFORE the probe runs, not +// after it is read: if the probe fails to start, crashes, or is still running +// when we look, an old file left on disk would be indistinguishable from a fresh +// result and we would report yesterday's mapped drives as current. Deleting up +// front means a file that exists at read time was necessarily produced by this +// cycle. +public sealed class UserContextSpool +{ + public const string SpoolFolderName = "ShopDB"; + public const string ResultFileName = "usercontext.json"; + + // A probe result is small - a handful of mapped drives and listeners. A file + // larger than this is either corrupt or hostile, and we will not parse it. + private const long MaxResultBytes = 512 * 1024; + + private readonly ILogger logger; + + public UserContextSpool(ILogger logger) + { + this.logger = logger; + } + + // Full path to the spool file for the signed-in user, or null (with a + // reason) when nobody is signed in. No session means no session-scoped data + // to collect, which is a normal state on an unattended machine, not a fault. + public string? TryGetResultPath(out string? reason) + { + var profileDirectory = + InteractiveProcessLauncher.TryGetActiveUserProfileDirectory(out var error); + if (string.IsNullOrWhiteSpace(profileDirectory)) + { + reason = error ?? "The interactive user profile could not be resolved."; + return null; + } + + reason = null; + return Path.Combine(profileDirectory, SpoolFolderName, ResultFileName); + } + + // Remove any result file left over from a previous cycle. Call this before + // launching the probe so a stale file can never be mistaken for a fresh one. + public void ClearStaleResult(string resultPath) + { + try + { + if (File.Exists(resultPath)) + { + File.Delete(resultPath); + logger.LogDebug( + "Removed a stale user-context spool file at {Path}", resultPath); + } + } + catch (Exception exception) + { + // Not fatal on its own, but it does mean we cannot trust whatever we + // read next, so the caller must know. + logger.LogWarning(exception, + "Could not remove the stale user-context spool file at {Path}", resultPath); + } + } + + public void EnsureSpoolDirectory(string resultPath) + { + var directory = Path.GetDirectoryName(resultPath); + if (string.IsNullOrWhiteSpace(directory)) + return; + + try + { + Directory.CreateDirectory(directory); + } + catch (Exception exception) + { + logger.LogWarning(exception, + "Could not create the user-context spool directory at {Directory}", directory); + } + } + + // Read and consume the probe result. Returns null when the probe produced + // nothing, which the caller reports as "no session data this cycle" rather + // than as empty data - absent and empty mean different things here. + public Dictionary? TryConsumeResult(string resultPath) + { + try + { + if (!File.Exists(resultPath)) + return null; + + var info = new FileInfo(resultPath); + if (info.Length == 0) + { + logger.LogDebug("The user-context spool file was empty."); + return null; + } + + if (info.Length > MaxResultBytes) + { + logger.LogWarning( + "The user-context spool file was {Bytes} bytes, above the {Limit} byte " + + "limit; discarding it.", + info.Length, MaxResultBytes); + return null; + } + + var json = File.ReadAllText(resultPath); + if (string.IsNullOrWhiteSpace(json)) + return null; + + var parsed = JsonSerializer.Deserialize>(json); + if (parsed is null || parsed.Count == 0) + return null; + + return parsed; + } + catch (JsonException exception) + { + // A truncated file is the expected shape of this failure: the probe + // was still writing when we looked. Next cycle gets a clean one. + logger.LogWarning(exception, + "The user-context spool file was not valid JSON; discarding it."); + return null; + } + catch (Exception exception) + { + logger.LogWarning(exception, + "Could not read the user-context spool file at {Path}", resultPath); + return null; + } + finally + { + ClearStaleResult(resultPath); + } + } +} diff --git a/agent/ShopDbEndpointAgent/Program.cs b/agent/ShopDbEndpointAgent/Program.cs index b003d9f6..efb1d67c 100644 --- a/agent/ShopDbEndpointAgent/Program.cs +++ b/agent/ShopDbEndpointAgent/Program.cs @@ -191,10 +191,28 @@ static void RegisterCoreServices(IServiceCollection services, AgentConfig config sp.GetRequiredService>())); } + // Persisted per-channel event watermark so each Windows event record is + // reported exactly once, instead of re-sending the newest window of every + // channel on every cadence tick. + services.AddSingleton(sp => new EventWatermarkStore( + sp.GetRequiredService>())); + + // Session-scoped collection. The agent is LocalSystem, so mapped drives and + // user-owned listeners are invisible to it; the probe runs a short script in + // the operator's session and hands the result back through a spool file. + services.AddSingleton(sp => new UserContextSpool( + sp.GetRequiredService>())); + + services.AddSingleton(sp => new UserContextProbe( + sp.GetRequiredService(), + sp.GetRequiredService>())); + services.AddSingleton(sp => new EndpointCollector( sp.GetRequiredService>(), config, - sp.GetService())); + sp.GetService(), + sp.GetRequiredService(), + sp.GetRequiredService())); services.AddSingleton(sp => new CollectorClient( config, sp.GetRequiredService>())); diff --git a/agent/ShopDbEndpointAgent/ShopDbEndpointAgent.csproj b/agent/ShopDbEndpointAgent/ShopDbEndpointAgent.csproj index 63e33106..a4bb5b59 100644 --- a/agent/ShopDbEndpointAgent/ShopDbEndpointAgent.csproj +++ b/agent/ShopDbEndpointAgent/ShopDbEndpointAgent.csproj @@ -7,7 +7,19 @@ enable ShopDbEndpointAgent shopdbagent - - 1.6.3 + 1.7.0 true win-x64 true diff --git a/plugins/endpoints/AGENT-CONTRACT.md b/plugins/endpoints/AGENT-CONTRACT.md index ff46608c..348f72f9 100644 --- a/plugins/endpoints/AGENT-CONTRACT.md +++ b/plugins/endpoints/AGENT-CONTRACT.md @@ -51,9 +51,44 @@ arrays. `adopt` is first-contact only. } ], "scheduledtasks": [ /* opaque array */ ], + "disks": [ + { + "drive": "C:", + "label": "OS", + "filesystem": "NTFS", + "sizebytes": 511000000000, + "freebytes": 128000000000, + "usedbytes": 383000000000, + "freepercent": 25.0 // precomputed; a threshold is a direct compare + } + ], + "bios": { /* opaque object: vendor, version, release date */ }, + "memory": { /* opaque object: totalbytes, freebytes */ }, + "processor": { /* opaque object: name, cores, logicalprocessors, maxclockmhz */ }, + "osinfo": { /* opaque object: build, version, install date */ }, + "domain": "shopfloor.example.com", // domain or workgroup membership + "pendingreboot": false, // omit when unknown; absent != false + "usercontext": { // omitted entirely when nobody is signed in + "netuse": [ + { + "localname": "Z:", + "remotename": "\\\\fileserver\\programs", + "status": "OK", + "username": "SHOPFLOOR\\operator", + "persistent": true + } + ], + "userlisteners": [ + { "localaddress": "0.0.0.0", "localport": 7878, "processid": 4512, "processname": "DncMain" } + ], + "idleseconds": 45, + "username": "operator", + "collectedat": "2025-01-14T06:12:00Z" + }, "events": [ { "eventid": 6008, // required per item; non-catalog IDs are dropped, not errored + "eventrecordid": 184412, // Windows record id; dedupe key, see notes "eventchannel": "System", "eventsource": "EventLog", "eventtime": "2025-01-14T06:11:58Z", @@ -81,6 +116,31 @@ Notes: (`HKLM\SOFTWARE\[WOW6432Node\]GE Aircraft Engines\DNC\General:MachineNo`) or the UDC `udc_settings_*.json` (`GeneralSettings.MachineNumber`), in that order (DESIGN-DECISIONS-agent.md 9.4/9.5). `""` or `"9999"` means none. +- `eventrecordid` is the Windows event record id: monotonic and unique within a + channel on a machine. The agent uses it as a per-channel harvest watermark and + the server enforces a unique index on + `(assetid, eventchannel, eventrecordid)`, so one real event is stored exactly + once. It is nullable: an older agent omits it, and MySQL permits repeated + NULLs in a unique index, so those reports keep the previous append behavior + rather than failing ingest. Send it whenever the source record exposes one. +- `disks`/`bios`/`memory`/`processor`/`osinfo`/`usercontext` are stored verbatim + as JSON snapshots on the endpoint state, newest-wins. `domain` and + `pendingreboot` are scalar columns because the fleet view filters and alerts + on them. +- OMITTING a key is not the same as sending an empty value. Every one of these + fields is applied only when the key is PRESENT, so a report that omits it + preserves the last-known value. Send `null`/`[]` only when you mean "this is + genuinely empty now"; omit when you simply did not collect it this cycle. +- `pendingreboot` follows that rule strictly: omit it when the check failed. A + literal `false` asserts the endpoint is confirmed NOT holding a reboot. +- `usercontext` is gathered by a probe the agent runs inside the signed-in + user's session, because mapped drives, user-owned listeners and idle time do + not exist in the agent's own LocalSystem session 0. It is omitted when no one + is signed in, which is a normal unattended state, not a fault. +- `ports` carries LISTENING sockets only. The full connection table is available + on demand via the `network_connections` command instead, because established + connections churn constantly and would change every report for no diagnostic + gain. ## 3. Response @@ -137,13 +197,23 @@ public sealed class EndpointReport [JsonPropertyName("ports")] public object[]? Ports { get; set; } [JsonPropertyName("interfaces")] public object[]? Interfaces { get; set; } [JsonPropertyName("scheduledtasks")]public object[]? ScheduledTasks { get; set; } + [JsonPropertyName("disks")] public object[]? Disks { get; set; } + [JsonPropertyName("bios")] public object? Bios { get; set; } + [JsonPropertyName("memory")] public object? Memory { get; set; } + [JsonPropertyName("processor")] public object? Processor { get; set; } + [JsonPropertyName("osinfo")] public object? OsInfo { get; set; } + [JsonPropertyName("domain")] public string? Domain { get; set; } + [JsonPropertyName("pendingreboot")] public bool? PendingReboot { get; set; } + [JsonPropertyName("usercontext")] public object? UserContext { get; set; } + [JsonPropertyName("applications")] public object[]? Applications { get; set; } [JsonPropertyName("events")] public EndpointEventRecord[]? Events { get; set; } } public sealed class EndpointEventRecord { - [JsonPropertyName("eventid")] public int EventId { get; set; } - [JsonPropertyName("eventchannel")] public string? EventChannel { get; set; } + [JsonPropertyName("eventid")] public int EventId { get; set; } + [JsonPropertyName("eventrecordid")] public long? EventRecordId { get; set; } + [JsonPropertyName("eventchannel")] public string? EventChannel { get; set; } [JsonPropertyName("eventsource")] public string? EventSource { get; set; } [JsonPropertyName("eventtime")] public DateTimeOffset? EventTime { get; set; } [JsonPropertyName("eventmessage")] public string? EventMessage { get; set; } @@ -230,6 +300,11 @@ Allowlisted actions (server source of truth: `plugins/endpoints/commands.py`): (`C:\ProgramData\UDC\MTConnect_UDC\Agent\run.bat`). - `mtconnect_stop` - stop the running MTConnect UDC agent process. - `mtconnect_status` - report whether the MTConnect UDC agent is running. +- `network_connections` - return the full TCP connection table with owning + process ids and names. Read-only diagnostic. The result is capped at 500 + connections with a `truncated` flag in `detail`, because an oversized payload + rejected by the server returns nothing at all and a partial answer is more + useful than a failed one. Standard reports carry listening ports only. ### 7.1 Claim the next command (long-poll) diff --git a/plugins/endpoints/commands.py b/plugins/endpoints/commands.py index 31b05376..2e2746e2 100644 --- a/plugins/endpoints/commands.py +++ b/plugins/endpoints/commands.py @@ -77,6 +77,12 @@ COMMAND_ALLOWLIST = { 'label': 'Collect now', 'description': 'Force the endpoint agent to report fresh telemetry off-cadence.', }, + 'network_connections': { + 'label': 'Show network connections', + 'description': 'Return the full TCP connection table with owning processes. ' + 'Read-only diagnostic; the standard report carries only ' + 'listening ports because established connections churn.', + }, 'run_script': { 'label': 'Run script', 'description': 'Run a ShopDB-hosted script on the endpoint. The agent ' diff --git a/plugins/endpoints/eventcatalog.py b/plugins/endpoints/eventcatalog.py index 5f6c8903..d382eeee 100644 --- a/plugins/endpoints/eventcatalog.py +++ b/plugins/endpoints/eventcatalog.py @@ -24,7 +24,14 @@ EVENT_CATALOG = { 55: ('File system corruption', 'critical', 'System'), 7011: ('Service timeout', 'critical', 'System'), 10016: ('DCOM permission error', 'warning', 'System'), + # Windows Error Reporting trio on the Application channel. 1000 is the + # faulting-application record (app name, version, faulting module, exception + # code) and is the one that says what actually broke; 1001 is the WER bucket + # record Windows writes alongside it; 1002 is a hang rather than a crash, so + # it is a warning - the app is unresponsive, not gone. + 1000: ('Application error', 'critical', 'Application'), 1001: ('Application crash', 'critical', 'Application'), + 1002: ('Application hang', 'warning', 'Application'), } # The IDs a tech should see surfaced fleet-wide on the dashboard stability card. diff --git a/plugins/endpoints/migrations/versions/0012_endpoint_eventrecordid.py b/plugins/endpoints/migrations/versions/0012_endpoint_eventrecordid.py new file mode 100644 index 00000000..360c721f --- /dev/null +++ b/plugins/endpoints/migrations/versions/0012_endpoint_eventrecordid.py @@ -0,0 +1,80 @@ +"""Dedup curated Windows events on the Windows event record id. + +Adds endpointevents.eventrecordid and a unique index over +(assetid, eventchannel, eventrecordid). + +Why: the agent used to re-read the newest window of every curated channel on +each cadence tick, and this ingest appended those rows without a dedup check, so +one real event was stored once per report for as long as it stayed in that +window - up to 96 identical rows a day. The agent now carries a per-channel +watermark keyed on the Windows record id, which is monotonic and unique within a +channel on a machine; this index is the server-side guarantee that survives an +agent reinstall, a restored image, or two overlapping reports, none of which +preserve the agent's watermark. + +Nullable on purpose: an agent older than this field reports no record id, and +MySQL permits repeated NULLs inside a unique index, so those rows keep the +previous append-only behavior instead of failing ingest. + +Existing rows are left alone. They all predate the column and so carry NULL, +which cannot collide with the new index. Purging the historical duplicates is a +separate, deliberate data cleanup - not something a schema migration should do +irreversibly. + +Idempotent: both the column-add and the index-create check the live schema +first, so a db.create_all() site and a re-run both stay clean on strict MySQL 8. +""" +from alembic import op + +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'endpoints0012eventrecordid' +down_revision = 'endpoints0011trayversion' +branch_labels = None +depends_on = None + +INDEX_NAME = 'idx_endpointevents_assetrecord' + + +def _has_column(table_name, column_name): + from sqlalchemy import inspect + + inspector = inspect(op.get_bind()) + if not inspector.has_table(table_name): + return False + return any(col['name'] == column_name + for col in inspector.get_columns(table_name)) + + +def _has_index(table_name, index_name): + from sqlalchemy import inspect + + inspector = inspect(op.get_bind()) + if not inspector.has_table(table_name): + return False + return any(index['name'] == index_name + for index in inspector.get_indexes(table_name)) + + +def upgrade(): + if not _has_column('endpointevents', 'eventrecordid'): + op.add_column( + 'endpointevents', + sa.Column('eventrecordid', sa.BigInteger(), nullable=True)) + + if not _has_index('endpointevents', INDEX_NAME): + op.create_index( + INDEX_NAME, + 'endpointevents', + ['assetid', 'eventchannel', 'eventrecordid'], + unique=True) + + +def downgrade(): + if _has_index('endpointevents', INDEX_NAME): + op.drop_index(INDEX_NAME, table_name='endpointevents') + + if _has_column('endpointevents', 'eventrecordid'): + op.drop_column('endpointevents', 'eventrecordid') diff --git a/plugins/endpoints/migrations/versions/0013_endpoint_inventory_columns.py b/plugins/endpoints/migrations/versions/0013_endpoint_inventory_columns.py new file mode 100644 index 00000000..b9844574 --- /dev/null +++ b/plugins/endpoints/migrations/versions/0013_endpoint_inventory_columns.py @@ -0,0 +1,93 @@ +"""Persist the hardware, OS, and session inventory the agent already reports. + +Adds to endpointstates: disksjson, biosjson, memoryjson, processorjson, +osinfojson, usercontextjson, domain, and pendingreboot. + +Why now: the agent has been sending disks, bios, memory, processor, osinfo and +domain on every report for several versions. The ingest logged them as unknown +payload keys and threw them away, which was the intended staging step - the +shapes are now proven against real endpoint traffic, so they get columns. +usercontext and pendingreboot are new in this same change. + +Shape follows the existing servicesjson/portsjson convention: newest snapshot +per endpoint as MEDIUMTEXT, not one row per sample. The agent resends the full +structure every cycle and operators ask "what is in this PC now", not "what was +in it in March". History would be pure write amplification on a fleet-sized +table. + +domain and pendingreboot are real columns rather than JSON because they are +single values the fleet view filters, groups, and alerts on; burying a scalar in +a blob makes it unqueryable. + +All columns are nullable. Every existing row predates them, and there is no +correct backfill value - a machine whose reboot state has never been reported is +NOT the same as one confirmed to have no reboot pending, and inventing False +would silently create a fleet of "healthy" endpoints. + +Idempotent: each column-add checks the live schema first, so a db.create_all() +site and a re-run both stay clean on strict MySQL 8. +""" +from alembic import op + +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'endpoints0013inventorycolumns' +down_revision = 'endpoints0012eventrecordid' +branch_labels = None +depends_on = None + +TABLE_NAME = 'endpointstates' + +# MEDIUMTEXT. An installed-application or disk list on a busy engineering +# workstation comfortably exceeds the 64KB TEXT ceiling, and a silent truncation +# there would corrupt the JSON rather than merely shorten it. +_SNAPSHOT_COLUMNS = ( + 'disksjson', + 'biosjson', + 'memoryjson', + 'processorjson', + 'osinfojson', + 'usercontextjson', +) + + +def _has_column(table_name, column_name): + from sqlalchemy import inspect + + inspector = inspect(op.get_bind()) + if not inspector.has_table(table_name): + return False + return any(col['name'] == column_name + for col in inspector.get_columns(table_name)) + + +def upgrade(): + for column_name in _SNAPSHOT_COLUMNS: + if not _has_column(TABLE_NAME, column_name): + op.add_column( + TABLE_NAME, + sa.Column(column_name, sa.Text(length=16777215), nullable=True)) + + if not _has_column(TABLE_NAME, 'domain'): + op.add_column( + TABLE_NAME, + sa.Column('domain', sa.String(length=255), nullable=True)) + + if not _has_column(TABLE_NAME, 'pendingreboot'): + op.add_column( + TABLE_NAME, + sa.Column('pendingreboot', sa.Boolean(), nullable=True)) + + +def downgrade(): + if _has_column(TABLE_NAME, 'pendingreboot'): + op.drop_column(TABLE_NAME, 'pendingreboot') + + if _has_column(TABLE_NAME, 'domain'): + op.drop_column(TABLE_NAME, 'domain') + + for column_name in reversed(_SNAPSHOT_COLUMNS): + if _has_column(TABLE_NAME, column_name): + op.drop_column(TABLE_NAME, column_name) diff --git a/plugins/endpoints/models/endpoint.py b/plugins/endpoints/models/endpoint.py index fcd0d733..1113ea0f 100644 --- a/plugins/endpoints/models/endpoint.py +++ b/plugins/endpoints/models/endpoint.py @@ -80,6 +80,28 @@ class EndpointState(db.Model): interfacesjson = db.Column(db.Text(length=16777215), nullable=True) scheduledtasksjson = db.Column(db.Text(length=16777215), nullable=True) + # Hardware and OS inventory, same newest-snapshot rule as above. These + # arrived on the wire well before this and were logged as unknown keys, so + # the shapes are already proven against real endpoint traffic. + disksjson = db.Column(db.Text(length=16777215), nullable=True) + biosjson = db.Column(db.Text(length=16777215), nullable=True) + memoryjson = db.Column(db.Text(length=16777215), nullable=True) + processorjson = db.Column(db.Text(length=16777215), nullable=True) + osinfojson = db.Column(db.Text(length=16777215), nullable=True) + + # Session-scoped data collected by a probe running as the signed-in user + # (mapped drives, user-owned listeners). Null when nobody is signed in, + # which is a normal unattended state rather than a fault. + usercontextjson = db.Column(db.Text(length=16777215), nullable=True) + + # Windows domain or workgroup membership. A scalar, not JSON: it is a single + # value the fleet view filters and groups on. + domain = db.Column(db.String(255), nullable=True) + + # True when the endpoint is holding a reboot. Null = never reported, which + # is deliberately distinct from a confident False. + pendingreboot = db.Column(db.Boolean, nullable=True) + # Warn-only reconciler output, refreshed on every agent check-in in the same # commit as servicesjson. Each entry names a policy/shipped-attached monitor # the endpoint is expected to be running but the latest snapshot does not @@ -116,6 +138,8 @@ class EndpointState(db.Model): 'idleseconds': self.idleseconds, 'cpupercent': self.cpupercent, 'memorypercent': self.memorypercent, + 'domain': self.domain, + 'pendingreboot': self.pendingreboot, 'mtconnectrunning': self.mtconnectrunning, 'mtconnectcheckedat': _utciso(self.mtconnectcheckedat), 'lastreportedat': _utciso(self.lastreportedat), @@ -255,6 +279,16 @@ class EndpointEvent(db.Model): eventchannel = db.Column(db.String(50), nullable=True) eventsource = db.Column(db.String(255), nullable=True) + # The Windows event record id: monotonic and unique within its channel on a + # given machine. This is the dedup key. The agent also uses it as its + # harvest watermark, so in normal operation a record is sent once; the + # unique index below is what guarantees it even when an agent is reinstalled + # (losing its watermark) or two reports overlap. Nullable because an agent + # older than this field still reports, and MySQL permits repeated NULLs in a + # unique index, so those degrade to the previous append-only behavior rather + # than failing. + eventrecordid = db.Column(db.BigInteger, nullable=True) + # When the event happened on the endpoint (from the record), not when it # reached ShopDB. eventtime = db.Column(db.DateTime, nullable=True, index=True) @@ -269,6 +303,8 @@ class EndpointEvent(db.Model): __table_args__ = ( db.Index('idx_endpointevents_assetevent', 'assetid', 'eventtime'), + db.Index('idx_endpointevents_assetrecord', + 'assetid', 'eventchannel', 'eventrecordid', unique=True), ) def __repr__(self): @@ -279,6 +315,7 @@ class EndpointEvent(db.Model): 'endpointeventid': self.endpointeventid, 'assetid': self.assetid, 'eventid': self.eventid, + 'eventrecordid': self.eventrecordid, 'eventchannel': self.eventchannel, 'eventsource': self.eventsource, 'eventtime': _utciso(self.eventtime), diff --git a/tests/test_plugin_migrations.py b/tests/test_plugin_migrations.py index 856583e1..fe5ac73d 100644 --- a/tests/test_plugin_migrations.py +++ b/tests/test_plugin_migrations.py @@ -101,7 +101,9 @@ EXPECTED_HEAD_REVISION['warranty'] = 'warranty0002proof' # 0010 renames the shipped business policy off the originating site's # designator on installs that already stamped 0007 (ADR-015). # 0011 adds the trayversion cache column to endpointstates. -EXPECTED_HEAD_REVISION['endpoints'] = 'endpoints0011trayversion' +# 0012 adds endpointevents.eventrecordid plus the unique index that dedups a +# replayed Windows event record. +EXPECTED_HEAD_REVISION['endpoints'] = 'endpoints0013inventorycolumns' # Plugins built after the cutover: their 0001 baseline really creates tables the # core chain never owned.