Fix user-context collection, split agent log destinations, widen upload cap
Mapped drives stopped reporting because the user-context path failed silently in three separate places, and none of them left a trace to read back. - Probe: launch via -EncodedCommand instead of staging a temp .ps1 and running powershell.exe -File. A missing or quarantined script file fails BEFORE any statement runs, so the script's own try/catch never sees it and the only symptom is exit -196608. - Spool: pin to the ShopFloor profile. Resolving it through WTSQueryUserToken is privileged, endpoint protection blocks it, and on failure the tray (writer) and service (reader) silently disagree about the path. - Spool: consume after a successful parse, so a stopped tray cannot keep re-reporting stale mapped drives as current. Unparseable files are left in place for the next cycle. - Logging: user-session components cannot write to Program Files or to the SYSTEM-owned ProgramData log directory, and the file sink degrades to a silent no-op when it cannot create its directory. Tray and probe now log to C:\Users\ShopFloor\ShopDB\Logs, installs transcript to ProgramData, and the sink records a SinkProblem instead of swallowing UnauthorizedAccessException. - Installer: provision the operator spool/log directories with ShopFloor:Modify before the tray first runs, and stop the transcript via trap so a throw cannot orphan exactly the failed install worth reading. - Upload cap 64 -> 128 MB: the self-contained tray publishes at ~69 MB and was rejected on SIZE, but the size check returns before the PE is parsed, so the operator only ever saw a version-detection failure. AGENT-CONTRACT.md gains a normative section defining the three security contexts, their log destinations, the fixed spool path and consume-on-read. Agent 1.8.0 -> 1.8.1, tray 1.1.0 -> 1.1.1 so the deployed build is identifiable from a log line.
This commit is contained in:
@@ -14,6 +14,8 @@ reaches the version gate.
|
||||
import io
|
||||
|
||||
from plugins.endpoints.api import agentstore
|
||||
from plugins.endpoints.api import agentupload
|
||||
from shopdb.config import Config
|
||||
|
||||
UPLOAD_URL = '/api/endpoints/siteagent/binaries/{}'
|
||||
|
||||
@@ -63,3 +65,26 @@ def test_upload_rejects_when_neither_typed_nor_pe_is_usable(
|
||||
response = _upload(client, auth_headers, monkeypatch, None)
|
||||
assert response.status_code == 400
|
||||
assert 'dotted-numeric' in response.get_json()['data']['error']['message']
|
||||
|
||||
|
||||
def test_upload_cap_admits_a_self_contained_single_file_publish():
|
||||
"""The tray publishes ~69 MB; the cap must clear a real binary.
|
||||
|
||||
This is the bug that presented as "upload does not detect version": the
|
||||
size check returns before the PE is parsed, so an oversized file looks
|
||||
like a parser failure to the operator. Pinned at 100 MB rather than the
|
||||
observed 69 MB so runtime growth does not silently re-break it.
|
||||
"""
|
||||
assert agentupload._MAX_UPLOAD_BYTES >= 100 * 1024 * 1024
|
||||
|
||||
|
||||
def test_request_cap_stays_above_the_agent_upload_cap():
|
||||
"""MAX_CONTENT_LENGTH bounds the whole multipart envelope, not the file.
|
||||
|
||||
Setting the two equal rejects a file sitting exactly at the agent cap,
|
||||
because the boundaries and form fields push the request over. They must
|
||||
move together, with real headroom between them.
|
||||
"""
|
||||
envelope = Config.MAX_CONTENT_LENGTH
|
||||
assert envelope > agentupload._MAX_UPLOAD_BYTES
|
||||
assert envelope - agentupload._MAX_UPLOAD_BYTES >= 1024 * 1024
|
||||
|
||||
148
tests/test_plugins/test_endpoints_ip_history.py
Normal file
148
tests/test_plugins/test_endpoints_ip_history.py
Normal file
@@ -0,0 +1,148 @@
|
||||
"""IP History and IP Configuration, carved out of the old Details Network card.
|
||||
|
||||
These read core Communication rows, which accumulate as an ACCIDENTAL history:
|
||||
_record_interfaces matches an incoming NIC on MAC then on IP and inserts when
|
||||
neither matches, and never deletes. So an address the machine has not held for
|
||||
a year sits next to the current one. These tests pin the reading of that, not
|
||||
the collecting.
|
||||
|
||||
The date columns are the delicate part. modifieddate only moves when a field on
|
||||
the row actually changes, so an address still held and reported unchanged keeps
|
||||
an old stamp. It is labelled 'Last changed' for that reason and must not drift
|
||||
back to 'Last seen'.
|
||||
"""
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from plugins.endpoints.api.routes import (
|
||||
_IPCONFIG_COLUMNS,
|
||||
_IPHISTORY_COLUMNS,
|
||||
_iphistory_status,
|
||||
_isostamp,
|
||||
)
|
||||
|
||||
|
||||
class FakeComm:
|
||||
"""Only the columns these two routes read."""
|
||||
|
||||
def __init__(self, ipaddress='10.0.0.1', macaddress='AA:BB',
|
||||
isprimary=False, ismachinenetwork=False, isdhcp=False,
|
||||
subnetmask='', gateway='', dns1='', dns2='',
|
||||
createddate=None, modifieddate=None):
|
||||
self.ipaddress = ipaddress
|
||||
self.macaddress = macaddress
|
||||
self.isprimary = isprimary
|
||||
self.ismachinenetwork = ismachinenetwork
|
||||
self.isdhcp = isdhcp
|
||||
self.subnetmask = subnetmask
|
||||
self.gateway = gateway
|
||||
self.dns1 = dns1
|
||||
self.dns2 = dns2
|
||||
self.createddate = createddate
|
||||
self.modifieddate = modifieddate
|
||||
|
||||
|
||||
def test_an_apipa_address_is_named_as_a_dhcp_failure():
|
||||
# 169.254.x.x is not an address the machine was given - it is what Windows
|
||||
# assigns when DHCP failed. Calling it 'Previous' would file a past network
|
||||
# fault as a past subnet.
|
||||
assert _iphistory_status(FakeComm(ipaddress='169.254.122.88')) == 'APIPA (no DHCP)'
|
||||
|
||||
|
||||
def test_apipa_wins_over_primary():
|
||||
# A box currently failing DHCP can have its APIPA address flagged primary
|
||||
# by the collector. The fault is the more useful thing to say.
|
||||
comm = FakeComm(ipaddress='169.254.1.1', isprimary=True)
|
||||
assert _iphistory_status(comm) == 'APIPA (no DHCP)'
|
||||
|
||||
|
||||
def test_a_normal_address_is_not_mistaken_for_apipa():
|
||||
# Guards a substring match: 169.254 must be a prefix, not found anywhere.
|
||||
assert _iphistory_status(FakeComm(ipaddress='10.169.254.5')) == 'Previous'
|
||||
|
||||
|
||||
def test_the_primary_address_is_marked():
|
||||
assert _iphistory_status(FakeComm(isprimary=True)) == 'Primary'
|
||||
|
||||
|
||||
def test_a_machine_network_address_is_distinguished_from_a_stale_one():
|
||||
# ismachinenetwork means it talks to the machine, not that it is old.
|
||||
assert _iphistory_status(FakeComm(ismachinenetwork=True)) == 'Machine network'
|
||||
|
||||
|
||||
def test_everything_else_reads_as_previous():
|
||||
assert _iphistory_status(FakeComm()) == 'Previous'
|
||||
|
||||
|
||||
def test_a_missing_timestamp_yields_null_not_a_crash():
|
||||
# createddate is nullable on legacy rows imported before BaseModel stamps.
|
||||
assert _isostamp(None) is None
|
||||
|
||||
|
||||
def test_a_timestamp_is_emitted_as_utc_iso():
|
||||
# Matches every other timestamp this file emits; the client formats it to
|
||||
# the viewer's locale rather than being pinned to the server's.
|
||||
stamp = _isostamp(datetime(2026, 9, 5, 11, 30, 0))
|
||||
assert stamp == '2026-09-05T11:30:00Z'
|
||||
|
||||
|
||||
def test_the_second_date_is_not_labelled_last_seen():
|
||||
# The whole reason for the label. modifieddate cannot support 'last seen':
|
||||
# an unchanged address reported daily keeps a months-old stamp.
|
||||
labels = {col['key']: col['label'] for col in _IPHISTORY_COLUMNS}
|
||||
assert labels['lastchanged'] == 'Last changed'
|
||||
assert 'seen' not in labels['lastchanged'].lower()
|
||||
assert labels['firstseen'] == 'First seen'
|
||||
|
||||
|
||||
def test_the_two_cards_do_not_duplicate_columns_beyond_the_address():
|
||||
# They are split so neither is eleven columns wide. The address is the
|
||||
# deliberate join between them; anything else repeated means the split
|
||||
# drifted back together.
|
||||
history = {col['key'] for col in _IPHISTORY_COLUMNS}
|
||||
config = {col['key'] for col in _IPCONFIG_COLUMNS}
|
||||
assert history & config == {'ipaddress'}
|
||||
|
||||
|
||||
def test_the_history_ordering_compiles_on_mysql():
|
||||
# Regression: the first cut ordered with .nullslast(), which is valid
|
||||
# ANSI/Postgres but blows up on MySQL with a 1064 syntax error. The test
|
||||
# suite runs on SQLite, so it passed and the page still 500'd in the
|
||||
# browser. Compile the real ordering against the MySQL dialect so the
|
||||
# dialect gap is caught here instead of in a container log.
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.dialects import mysql
|
||||
|
||||
from shopdb.api import Communication
|
||||
|
||||
ordering = (select(Communication.ipaddress)
|
||||
.where(Communication.assetid == 1)
|
||||
.order_by(Communication.modifieddate.is_(None),
|
||||
Communication.modifieddate.desc(),
|
||||
Communication.communicationid.desc()))
|
||||
|
||||
compiled = str(ordering.compile(dialect=mysql.dialect()))
|
||||
assert 'NULLS LAST' not in compiled.upper()
|
||||
|
||||
|
||||
def test_every_panel_handler_still_has_a_route():
|
||||
# Regression: inserting the ipconfig route directly above panel_applications
|
||||
# ate that function's @endpoints_bp.route line, leaving a handler decorated
|
||||
# with require_permission only. It imported fine and every test passed; the
|
||||
# endpoint was simply gone. Assert each panel_* handler is actually mapped.
|
||||
import re
|
||||
|
||||
from plugins.endpoints.api import routes as routes_module
|
||||
|
||||
source = Path(routes_module.__file__).read_text(encoding='utf-8')
|
||||
handlers = re.findall(r'^def (panel_\w+)', source, re.MULTILINE)
|
||||
assert handlers, 'expected panel handlers in routes.py'
|
||||
|
||||
routed = set()
|
||||
for match in re.finditer(r'^@endpoints_bp\.route\(', source, re.MULTILINE):
|
||||
tail = source[match.start():]
|
||||
found = re.search(r'^def (\w+)', tail, re.MULTILINE)
|
||||
if found:
|
||||
routed.add(found.group(1))
|
||||
|
||||
assert set(handlers) - routed == set()
|
||||
@@ -69,7 +69,7 @@ def test_endpoints_asset_panels_are_computer_scoped():
|
||||
"""Every asset panel targets computer assets only, with an endpoint."""
|
||||
plugin = EndpointsPlugin()
|
||||
panels = plugin.get_asset_panels()
|
||||
assert len(panels) == 13
|
||||
assert len(panels) == 15
|
||||
for panel in panels:
|
||||
assert panel['assettypes'] == ['computer']
|
||||
assert panel['endpoint'].startswith('/api/endpoints/')
|
||||
|
||||
Reference in New Issue
Block a user