Merge the wilm endpoint-monitor and channel-distribution work into main
Six commits from wilm-import: the squashed import, route documentation, the channel remap onto our tag-and-trunk model, the project map, parallel test support, and the four smaller review findings. Merged rather than cherry-picked: the published mirror rewrites hashes, so the commits do not apply one at a time against our history even though the squashed diff applies clean. Scope: product # Conflicts: # docs/PROJECT-MAP.md
This commit is contained in:
cproudlock
@@ -98,7 +98,15 @@ EXEMPT_ENDPOINTS = {'knowledgebase.track_click', 'users.update_user',
|
||||
# kill switch. Enqueue (WebUI -> operator) IS permission-gated
|
||||
# on endpoints.command and IS swept. Contract in
|
||||
# plugins/endpoints/AGENT-CONTRACT.md section 7.
|
||||
'endpoints.report_command_result'}
|
||||
'endpoints.report_command_result',
|
||||
# Agent-facing operator issue report upload. Authenticated by
|
||||
# the endpoints collector API key (same shape as the exempt
|
||||
# collector blueprint), not JWT+role: the tray hands the
|
||||
# screenshot and note to the service, which uploads on the
|
||||
# collector key it already holds. The human-facing read and
|
||||
# image-serve routes ARE permission-gated on endpoints.view
|
||||
# and ARE swept.
|
||||
'endpoints.upload_operator_report'}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
|
||||
@@ -96,6 +96,14 @@ PUBLIC_ENDPOINTS = {
|
||||
'endpoints.download_app_package',
|
||||
'endpoints.download_app_package_by_app',
|
||||
'endpoints.download_script_body',
|
||||
|
||||
# The endpoint agent fetches its server-resolved monitor watch-list (the
|
||||
# processes/services site policy attaches to its PC type) on each telemetry
|
||||
# cycle. Authenticated inside the handler by the endpoints collector API key
|
||||
# - the same shape as claim_next_command - because a bay has no user session.
|
||||
# It is a read of already-scoped policy output, not an authority: attaching a
|
||||
# monitor is gated on the policy editor, audited with the requesting user.
|
||||
'endpoints.resolved_monitors_for_agent',
|
||||
}
|
||||
|
||||
# Reads that need a LOGIN but no named permission, because the handler decides
|
||||
|
||||
@@ -92,9 +92,12 @@ EXPECTED_HEAD_REVISION['warranty'] = 'warranty0002proof'
|
||||
# endpoints is a NEW post-cutover plugin: its 0001 genuinely creates the
|
||||
# endpointstates + endpointevents tables the core chain never owned, 0002
|
||||
# adds the endpointcommands queue plus the MTConnect state columns, 0003
|
||||
# adds the generic command target/targetkind columns, and 0004 adds the
|
||||
# endpointapplications SBOM snapshot table.
|
||||
EXPECTED_HEAD_REVISION['endpoints'] = 'endpoints0005cmdpin'
|
||||
# adds the generic command target/targetkind columns, 0004 adds the
|
||||
# endpointapplications SBOM snapshot table, 0005 pins a command sha256, 0006
|
||||
# seeds the reusable monitor catalog, 0007 ships the shopfloor monitor policy
|
||||
# bundle, 0008 adds the warn-only reconciler drift cache column, and 0009 adds
|
||||
# the operatorreports table for tray issue reports.
|
||||
EXPECTED_HEAD_REVISION['endpoints'] = 'endpoints0009opreports'
|
||||
|
||||
# Plugins built after the cutover: their 0001 baseline really creates tables the
|
||||
# core chain never owned.
|
||||
|
||||
285
tests/test_plugins/test_endpoints_operator_reports.py
Normal file
285
tests/test_plugins/test_endpoints_operator_reports.py
Normal file
@@ -0,0 +1,285 @@
|
||||
"""Operator issue report upload, retention purge, and hardened serve.
|
||||
|
||||
The tray captures a screenshot and a free-text note; the agent posts them to
|
||||
POST /api/endpoints/operatorreports on the collector key. The bytes land on
|
||||
disk under the operator-reports uploads dir and the row holds metadata. The
|
||||
human-facing panel and image-serve routes are JWT + endpoints.view gated, and
|
||||
the serve route carries the inert-upload headers. A retention sweep purges rows
|
||||
and files older than the configured window.
|
||||
"""
|
||||
|
||||
import io
|
||||
import os
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
from shopdb.core.models import Asset, AssetType, Setting
|
||||
|
||||
|
||||
UPLOAD_URL = '/api/endpoints/operatorreports'
|
||||
PANEL_URL = '/api/endpoints/panel/{}/operatorreports'
|
||||
IMAGE_URL = '/api/endpoints/operatorreports/{}/image'
|
||||
COLLECTOR_KEY = 'testcollectorkey'
|
||||
|
||||
|
||||
def _collector_headers(app):
|
||||
app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
|
||||
return {'X-API-Key': COLLECTOR_KEY}
|
||||
|
||||
|
||||
def _computer(db, hostname):
|
||||
"""Create a computer Asset + Computer; return the asset."""
|
||||
from plugins.computers.models import Computer, ComputerType
|
||||
|
||||
atype = AssetType.query.filter_by(assettype='computer').first()
|
||||
if not atype:
|
||||
atype = AssetType(assettype='computer')
|
||||
db.session.add(atype)
|
||||
db.session.flush()
|
||||
|
||||
ctype = ComputerType.query.filter_by(computertype='CMM PC').first()
|
||||
if not ctype:
|
||||
ctype = ComputerType(computertype='CMM PC')
|
||||
db.session.add(ctype)
|
||||
db.session.flush()
|
||||
|
||||
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
|
||||
isactive=True)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
comp = Computer(assetid=asset.assetid, hostname=hostname,
|
||||
computertypeid=ctype.computertypeid)
|
||||
db.session.add(comp)
|
||||
db.session.commit()
|
||||
return asset
|
||||
|
||||
|
||||
def _png_bytes():
|
||||
"""A tiny valid 1x1 PNG."""
|
||||
import base64
|
||||
return base64.b64decode(
|
||||
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk'
|
||||
'YPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==')
|
||||
|
||||
|
||||
def test_upload_creates_report_and_stores_screenshot(client, db, app):
|
||||
"""A collector-key upload with a screenshot creates a row and a file."""
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
|
||||
asset = _computer(db, 'OPREPORT1')
|
||||
data = {
|
||||
'hostname': 'OPREPORT1',
|
||||
'reason': 'Other (free text)',
|
||||
'details': 'the CMM app will not start',
|
||||
'screenshot': (io.BytesIO(_png_bytes()), 'shot.png'),
|
||||
}
|
||||
response = client.post(UPLOAD_URL, data=data,
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 201
|
||||
body = response.get_json()['data']
|
||||
assert body['assetid'] == asset.assetid
|
||||
assert body['details'] == 'the CMM app will not start'
|
||||
assert body['hasimage'] is True
|
||||
|
||||
report = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
assert report is not None
|
||||
assert report.imagefilename
|
||||
stored = os.path.join(app.instance_path, 'operatorreports',
|
||||
report.imagefilename)
|
||||
assert os.path.isfile(stored)
|
||||
|
||||
|
||||
def test_upload_without_screenshot_is_allowed(client, db, app):
|
||||
"""A note-only report (no screenshot) is valid; hasimage is false."""
|
||||
_computer(db, 'OPREPORT2')
|
||||
response = client.post(
|
||||
UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT2', 'details': 'noisy fan'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 201
|
||||
assert response.get_json()['data']['hasimage'] is False
|
||||
|
||||
|
||||
def test_upload_unknown_hostname_is_validation_error(client, db, app):
|
||||
"""A hostname with no computer row is rejected, not silently stored."""
|
||||
response = client.post(
|
||||
UPLOAD_URL,
|
||||
data={'hostname': 'NOSUCHBOX', 'details': 'x'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_upload_requires_collector_auth(client, db, app):
|
||||
"""Without the collector key the upload refuses (no JWT fallback)."""
|
||||
_computer(db, 'OPREPORT3')
|
||||
app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
|
||||
response = client.post(
|
||||
UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT3', 'details': 'x'},
|
||||
content_type='multipart/form-data')
|
||||
assert response.status_code in (401, 403)
|
||||
|
||||
|
||||
def test_panel_lists_reports_newest_first(client, db, app, auth_headers):
|
||||
"""The panel route returns this asset's reports, newest first."""
|
||||
asset = _computer(db, 'OPREPORT4')
|
||||
for note in ('first', 'second'):
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT4', 'details': note},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
|
||||
response = client.get(PANEL_URL.format(asset.assetid), headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
reports = response.get_json()['data']['reports']
|
||||
assert [r['details'] for r in reports] == ['second', 'first']
|
||||
|
||||
|
||||
def test_served_image_is_hardened(client, db, app, auth_headers):
|
||||
"""The screenshot serve route carries the inert-upload headers."""
|
||||
from shopdb.utils.uploads import UPLOAD_CSP
|
||||
|
||||
asset = _computer(db, 'OPREPORT5')
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT5', 'details': 'x',
|
||||
'screenshot': (io.BytesIO(_png_bytes()), 'shot.png')},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
report = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
|
||||
response = client.get(IMAGE_URL.format(report.operatorreportid),
|
||||
headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
assert response.headers.get('Content-Security-Policy') == UPLOAD_CSP
|
||||
assert response.headers.get('X-Content-Type-Options') == 'nosniff'
|
||||
|
||||
|
||||
def test_image_serve_requires_jwt(client, db, app):
|
||||
"""The screenshot serve route is not reachable without a user JWT."""
|
||||
asset = _computer(db, 'OPREPORT6')
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT6', 'details': 'x',
|
||||
'screenshot': (io.BytesIO(_png_bytes()), 'shot.png')},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
report = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
|
||||
response = client.get(IMAGE_URL.format(report.operatorreportid))
|
||||
assert response.status_code in (401, 403)
|
||||
|
||||
|
||||
def test_retention_purge_removes_expired_reports_and_files(client, db, app):
|
||||
"""A report past the retention window is purged, file and all, on upload."""
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
from plugins.endpoints.api.routes import (
|
||||
OPERATOR_REPORT_RETENTION_SETTING,
|
||||
)
|
||||
|
||||
db.session.add(Setting(key=OPERATOR_REPORT_RETENTION_SETTING, value='7'))
|
||||
db.session.commit()
|
||||
|
||||
asset = _computer(db, 'OPREPORT7')
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT7', 'details': 'old',
|
||||
'screenshot': (io.BytesIO(_png_bytes()), 'shot.png')},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
old = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
old_file = os.path.join(app.instance_path, 'operatorreports',
|
||||
old.imagefilename)
|
||||
assert os.path.isfile(old_file)
|
||||
|
||||
# Backdate it past the window, then upload a fresh report to trigger the
|
||||
# opportunistic sweep.
|
||||
old.createdat = old.createdat - timedelta(days=8)
|
||||
db.session.commit()
|
||||
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT7', 'details': 'new'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
|
||||
assert db.session.get(OperatorReport, old.operatorreportid) is None
|
||||
assert not os.path.isfile(old_file)
|
||||
remaining = OperatorReport.query.filter_by(assetid=asset.assetid).all()
|
||||
assert [r.details for r in remaining] == ['new']
|
||||
|
||||
|
||||
def test_retention_disabled_keeps_old_reports(client, db, app):
|
||||
"""A retention of 0 disables the sweep; old rows survive."""
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
from plugins.endpoints.api.routes import (
|
||||
OPERATOR_REPORT_RETENTION_SETTING,
|
||||
)
|
||||
|
||||
db.session.add(Setting(key=OPERATOR_REPORT_RETENTION_SETTING, value='0'))
|
||||
db.session.commit()
|
||||
|
||||
asset = _computer(db, 'OPREPORT8')
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT8', 'details': 'old'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
old = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
old.createdat = old.createdat - timedelta(days=365)
|
||||
db.session.commit()
|
||||
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT8', 'details': 'new'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
|
||||
assert db.session.get(OperatorReport, old.operatorreportid) is not None
|
||||
|
||||
|
||||
# --- what the route refuses to store -----------------------------------------
|
||||
#
|
||||
# The stored file is served back as image/png, and the size cap is the only
|
||||
# thing bounding what an agent key can push into the instance directory. Both
|
||||
# guards used to be nominal: the cap was checked AFTER upload.read() had already
|
||||
# pulled the whole body into memory, and the extension was forced to .png
|
||||
# whatever the bytes were.
|
||||
|
||||
def test_upload_refuses_bytes_that_are_not_a_png(client, db, app):
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
|
||||
_computer(db, 'OPREPORT-NOTPNG')
|
||||
data = {
|
||||
'hostname': 'OPREPORT-NOTPNG',
|
||||
'details': 'not really a screenshot',
|
||||
'screenshot': (io.BytesIO(b'MZ\x90\x00 this is a PE, not an image'),
|
||||
'shot.png'),
|
||||
}
|
||||
response = client.post(UPLOAD_URL, data=data,
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 400
|
||||
# Nothing was written: not the row, and not a file under the uploads dir.
|
||||
assert OperatorReport.query.filter_by(
|
||||
reportedhostname='OPREPORT-NOTPNG').first() is None
|
||||
|
||||
|
||||
def test_upload_refuses_an_image_over_the_cap(client, db, app):
|
||||
from plugins.endpoints.api.routes import OPERATOR_REPORT_MAX_IMAGE_BYTES
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
|
||||
_computer(db, 'OPREPORT-HUGE')
|
||||
# A real PNG header so the size check is what rejects it, not the signature.
|
||||
oversized = _png_bytes() + b'\x00' * (OPERATOR_REPORT_MAX_IMAGE_BYTES + 1)
|
||||
data = {
|
||||
'hostname': 'OPREPORT-HUGE',
|
||||
'details': 'oversized',
|
||||
'screenshot': (io.BytesIO(oversized), 'shot.png'),
|
||||
}
|
||||
response = client.post(UPLOAD_URL, data=data,
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 400
|
||||
assert 'limit' in response.get_json()['data']['error']['message']
|
||||
assert OperatorReport.query.filter_by(
|
||||
reportedhostname='OPREPORT-HUGE').first() is None
|
||||
@@ -69,11 +69,12 @@ 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) == 6
|
||||
assert len(panels) == 7
|
||||
for panel in panels:
|
||||
assert panel['assettypes'] == ['computer']
|
||||
assert panel['endpoint'].startswith('/api/endpoints/asset/{assetid}')
|
||||
assert panel['render'] in ('keyvalue', 'list', 'table')
|
||||
assert panel['endpoint'].startswith('/api/endpoints/')
|
||||
assert '{assetid}' in panel['endpoint']
|
||||
assert panel['render'] in ('keyvalue', 'list', 'table', 'operatorreports')
|
||||
|
||||
|
||||
def test_services_and_processes_are_split_into_two_filtered_cards():
|
||||
@@ -234,7 +235,6 @@ def test_new_hardware_telemetry_keys_are_logged_not_silently_known():
|
||||
assert 'applications' in KNOWN_PAYLOAD_KEYS
|
||||
|
||||
|
||||
|
||||
def test_telemetrymeta_route_is_registered_for_freshness_gating():
|
||||
"""The WebUI gates panel refreshes on lastreportedat advancing, so the
|
||||
lightweight freshness probe must be exposed on the plugin blueprint."""
|
||||
@@ -249,3 +249,26 @@ def test_telemetrymeta_route_is_registered_for_freshness_gating():
|
||||
url_prefix='/api/endpoints')
|
||||
rules = {rule.rule for rule in application.url_map.iter_rules()}
|
||||
assert any(rule.endswith('/asset/<int:asset_id>/telemetrymeta') for rule in rules)
|
||||
|
||||
|
||||
def test_fleet_list_route_resolves_without_trailing_slash():
|
||||
"""The Fleet list route must match the slash-less URL the SPA calls.
|
||||
|
||||
The list is the only route on the bare blueprint prefix. If it demands a
|
||||
trailing slash, Flask 308-redirects /api/endpoints to /api/endpoints/, and
|
||||
behind the /shopdb reverse proxy that redirect drops the path prefix and
|
||||
the list call fails while the detail routes (which carry a path suffix)
|
||||
keep working. strict_slashes=False makes both forms resolve directly.
|
||||
"""
|
||||
from flask import Flask
|
||||
|
||||
application = Flask(__name__)
|
||||
application.register_blueprint(
|
||||
EndpointsPlugin().get_blueprint(), url_prefix='/api/endpoints')
|
||||
adapter = application.url_map.bind('localhost')
|
||||
|
||||
endpoint, _ = adapter.match('/api/endpoints', method='GET')
|
||||
assert endpoint.endswith('list_endpoints')
|
||||
endpoint_slash, _ = adapter.match('/api/endpoints/', method='GET')
|
||||
assert endpoint_slash.endswith('list_endpoints')
|
||||
|
||||
|
||||
199
tests/test_plugins/test_endpoints_policy_monitors.py
Normal file
199
tests/test_plugins/test_endpoints_policy_monitors.py
Normal file
@@ -0,0 +1,199 @@
|
||||
"""Policy-attached monitors surface in the Tracked Services/Processes panel.
|
||||
|
||||
A monitor attached to a PC type through an active Business Policy (over a scoped
|
||||
Configuration Policy) must appear in GET /api/endpoints/asset/<id>/services even
|
||||
before the agent has reported that process/service, so the PC Detail Monitors
|
||||
tab shows the intended coverage as an Unknown row.
|
||||
"""
|
||||
|
||||
from shopdb.core.models import Asset, AssetType, Setting
|
||||
|
||||
|
||||
SERVICES_URL = '/api/endpoints/asset/{}/services'
|
||||
|
||||
|
||||
def _computer(db, hostname, computertype):
|
||||
"""Create a computer Asset + Computer of the given PC type; return the asset."""
|
||||
from plugins.computers.models import Computer, ComputerType
|
||||
|
||||
atype = AssetType.query.filter_by(assettype='computer').first()
|
||||
if not atype:
|
||||
atype = AssetType(assettype='computer')
|
||||
db.session.add(atype)
|
||||
db.session.flush()
|
||||
|
||||
ctype = ComputerType.query.filter_by(computertype=computertype).first()
|
||||
if not ctype:
|
||||
ctype = ComputerType(computertype=computertype)
|
||||
db.session.add(ctype)
|
||||
db.session.flush()
|
||||
|
||||
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
|
||||
isactive=True)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
comp = Computer(assetid=asset.assetid, hostname=hostname,
|
||||
computertypeid=ctype.computertypeid)
|
||||
db.session.add(comp)
|
||||
db.session.flush()
|
||||
return asset, ctype
|
||||
|
||||
|
||||
def _attach_monitor_to_pctype(db, monitor, computertypeid):
|
||||
"""Create a Configuration Policy scoped to the PC type carrying the monitor,
|
||||
wrapped in an active Business Policy, with the feature flag enabled."""
|
||||
from plugins.computers.models import (
|
||||
ConfigurationPolicy, ConfigurationPolicyAction,
|
||||
BusinessPolicy, BusinessPolicyItem,
|
||||
)
|
||||
|
||||
db.session.add(Setting(key='computers_configpolicies_enabled', value='true'))
|
||||
|
||||
policy = ConfigurationPolicy(
|
||||
name='Edge coverage',
|
||||
scopejson={'type': 'pctype', 'id': computertypeid},
|
||||
)
|
||||
db.session.add(policy)
|
||||
db.session.flush()
|
||||
db.session.add(ConfigurationPolicyAction(
|
||||
configurationpolicyid=policy.configurationpolicyid,
|
||||
action=f'monitor:{monitor.endpointmonitorid}',
|
||||
mode='gate',
|
||||
))
|
||||
|
||||
business = BusinessPolicy(name='Test Policy', priority=100)
|
||||
db.session.add(business)
|
||||
db.session.flush()
|
||||
db.session.add(BusinessPolicyItem(
|
||||
businesspolicyid=business.businesspolicyid,
|
||||
configurationpolicyid=policy.configurationpolicyid,
|
||||
sortorder=0,
|
||||
))
|
||||
db.session.commit()
|
||||
|
||||
|
||||
def test_policy_attached_process_monitor_surfaces_as_unknown_row(
|
||||
client, db, auth_headers):
|
||||
"""A msedge.exe process monitor attached to a PC type by policy appears in
|
||||
the services panel as an Unknown process row, even with no agent report."""
|
||||
from plugins.endpoints.models import EndpointMonitor
|
||||
|
||||
asset, ctype = _computer(db, 'F18BCDB4', 'CMM PC')
|
||||
monitor = EndpointMonitor(
|
||||
kind='process', displayname='msedge.exe', processname='msedge.exe',
|
||||
source='local', isactive=True,
|
||||
)
|
||||
db.session.add(monitor)
|
||||
db.session.flush()
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
response = client.get(SERVICES_URL.format(asset.assetid), headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
rows = response.get_json()['data']
|
||||
|
||||
edge = next((r for r in rows if r['name'] == 'msedge.exe'), None)
|
||||
assert edge is not None, 'policy-attached monitor missing from panel'
|
||||
assert edge['kind'] == 'process'
|
||||
assert edge['statelabel'] == 'Unknown'
|
||||
assert edge['statecolor'] == 'gray'
|
||||
# Admin holds endpoints.command, so the row offers a Start control whose
|
||||
# wire target is the OS process name.
|
||||
assert edge['action']['target'] == 'msedge.exe'
|
||||
assert edge['action']['targetkind'] == 'process'
|
||||
assert edge['action']['running'] is False
|
||||
|
||||
|
||||
def test_flag_off_hides_policy_monitor_from_panel(client, db, auth_headers):
|
||||
"""With the feature flag off the panel returns only reported rows (none)."""
|
||||
from plugins.endpoints.models import EndpointMonitor
|
||||
|
||||
asset, ctype = _computer(db, 'F18BCDB5', 'CMM PC')
|
||||
monitor = EndpointMonitor(
|
||||
kind='process', displayname='msedge.exe', processname='msedge.exe',
|
||||
source='local', isactive=True,
|
||||
)
|
||||
db.session.add(monitor)
|
||||
db.session.flush()
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
# Flip the flag off after the helper turned it on.
|
||||
flag = Setting.query.filter_by(
|
||||
key='computers_configpolicies_enabled').first()
|
||||
flag.value = 'false'
|
||||
db.session.commit()
|
||||
|
||||
response = client.get(SERVICES_URL.format(asset.assetid), headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
rows = response.get_json()['data']
|
||||
assert not any(r['name'] == 'msedge.exe' for r in rows)
|
||||
|
||||
|
||||
RESOLVED_URL = '/api/endpoints/monitors/resolved'
|
||||
COLLECTOR_KEY = 'testcollectorkey'
|
||||
|
||||
|
||||
def _collector_headers(app):
|
||||
"""Authorize the agent-facing route on the shared collector key."""
|
||||
app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
|
||||
return {'X-API-Key': COLLECTOR_KEY}
|
||||
|
||||
|
||||
def test_resolved_monitors_returns_policy_process_for_agent(client, db, app):
|
||||
"""The agent watch-list route returns a policy-attached process monitor,
|
||||
shaped as processes[]/services[] of {name, app}, keyed by hostname."""
|
||||
from plugins.endpoints.models import EndpointMonitor
|
||||
|
||||
asset, ctype = _computer(db, 'AGENTPC1', 'CMM PC')
|
||||
monitor = EndpointMonitor(
|
||||
kind='process', displayname='Edge', processname='msedge.exe',
|
||||
source='local', isactive=True,
|
||||
)
|
||||
db.session.add(monitor)
|
||||
db.session.flush()
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
response = client.get(RESOLVED_URL, query_string={'hostname': 'AGENTPC1'},
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()['data']
|
||||
assert {'name': 'msedge.exe', 'app': 'Edge'} in data['processes']
|
||||
assert data['services'] == []
|
||||
|
||||
|
||||
def test_resolved_monitors_carries_launchpath_for_process(client, db, app):
|
||||
"""A process monitor with a launch path set surfaces launchpath on the wire
|
||||
so the agent can honor a site-provisioned Start path (after validation)."""
|
||||
from plugins.endpoints.models import EndpointMonitor
|
||||
|
||||
asset, ctype = _computer(db, 'AGENTPC3', 'CMM PC')
|
||||
monitor = EndpointMonitor(
|
||||
kind='process', displayname='Edge', processname='msedge.exe',
|
||||
launchpath=r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe',
|
||||
source='local', isactive=True,
|
||||
)
|
||||
db.session.add(monitor)
|
||||
db.session.flush()
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
response = client.get(RESOLVED_URL, query_string={'hostname': 'AGENTPC3'},
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()['data']
|
||||
edge = next((p for p in data['processes'] if p['name'] == 'msedge.exe'), None)
|
||||
assert edge is not None
|
||||
assert edge['launchpath'] == \
|
||||
r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe'
|
||||
|
||||
|
||||
def test_resolved_monitors_requires_collector_auth(client, db, app):
|
||||
"""Without the collector key the agent route refuses (no JWT fallback)."""
|
||||
_computer(db, 'AGENTPC2', 'CMM PC')
|
||||
app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
|
||||
response = client.get(RESOLVED_URL, query_string={'hostname': 'AGENTPC2'})
|
||||
assert response.status_code in (401, 403)
|
||||
|
||||
|
||||
def test_resolved_monitors_unknown_hostname_is_404(client, db, app):
|
||||
"""A hostname with no computer row is a 404, not an empty 200."""
|
||||
response = client.get(RESOLVED_URL, query_string={'hostname': 'NOSUCHBOX'},
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 404
|
||||
238
tests/test_plugins/test_endpoints_reconciler.py
Normal file
238
tests/test_plugins/test_endpoints_reconciler.py
Normal file
@@ -0,0 +1,238 @@
|
||||
"""Warn-only reconciler: drift computation and its exposure on the panels.
|
||||
|
||||
The reconciler runs server-side on agent check-in. For every monitor a site
|
||||
policy attaches to an endpoint's PC type that is EXPECTED to be running, it flags
|
||||
DRIFT when the latest snapshot does not report it running. It never starts or
|
||||
stops anything - drift is surfaced for an operator to act on manually.
|
||||
|
||||
These tests cover the pure computer (compute_drift), the check-in write path
|
||||
(apply_collector_payload persisting driftjson), the panel exposure, and the
|
||||
observe-only seam (monitor_expects_running) that keeps a non-keepalive monitor
|
||||
out of the drift list.
|
||||
"""
|
||||
|
||||
from shopdb.core.models import Asset, AssetType, Setting
|
||||
|
||||
|
||||
SERVICES_URL = '/api/endpoints/asset/{}/services'
|
||||
STATUS_URL = '/api/endpoints/asset/{}/status'
|
||||
|
||||
|
||||
def _computer(db, hostname, computertype):
|
||||
"""Create a computer Asset + Computer of the given PC type; return the pair."""
|
||||
from plugins.computers.models import Computer, ComputerType
|
||||
|
||||
atype = AssetType.query.filter_by(assettype='computer').first()
|
||||
if not atype:
|
||||
atype = AssetType(assettype='computer')
|
||||
db.session.add(atype)
|
||||
db.session.flush()
|
||||
|
||||
ctype = ComputerType.query.filter_by(computertype=computertype).first()
|
||||
if not ctype:
|
||||
ctype = ComputerType(computertype=computertype)
|
||||
db.session.add(ctype)
|
||||
db.session.flush()
|
||||
|
||||
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
|
||||
isactive=True)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
comp = Computer(assetid=asset.assetid, hostname=hostname,
|
||||
computertypeid=ctype.computertypeid)
|
||||
db.session.add(comp)
|
||||
db.session.flush()
|
||||
return asset, ctype
|
||||
|
||||
|
||||
def _attach_monitor_to_pctype(db, monitor, computertypeid):
|
||||
"""Attach a monitor to a PC type via an active Business/Configuration Policy."""
|
||||
from plugins.computers.models import (
|
||||
ConfigurationPolicy, ConfigurationPolicyAction,
|
||||
BusinessPolicy, BusinessPolicyItem,
|
||||
)
|
||||
|
||||
if not Setting.query.filter_by(
|
||||
key='computers_configpolicies_enabled').first():
|
||||
db.session.add(
|
||||
Setting(key='computers_configpolicies_enabled', value='true'))
|
||||
|
||||
policy = ConfigurationPolicy(
|
||||
name='Coverage {}'.format(monitor.endpointmonitorid),
|
||||
scopejson={'type': 'pctype', 'id': computertypeid},
|
||||
)
|
||||
db.session.add(policy)
|
||||
db.session.flush()
|
||||
db.session.add(ConfigurationPolicyAction(
|
||||
configurationpolicyid=policy.configurationpolicyid,
|
||||
action='monitor:{}'.format(monitor.endpointmonitorid),
|
||||
mode='gate',
|
||||
))
|
||||
business = BusinessPolicy(name='BP {}'.format(monitor.endpointmonitorid),
|
||||
priority=100)
|
||||
db.session.add(business)
|
||||
db.session.flush()
|
||||
db.session.add(BusinessPolicyItem(
|
||||
businesspolicyid=business.businesspolicyid,
|
||||
configurationpolicyid=policy.configurationpolicyid,
|
||||
sortorder=0,
|
||||
))
|
||||
db.session.commit()
|
||||
|
||||
|
||||
def _process_monitor(db, displayname, processname):
|
||||
from plugins.endpoints.models import EndpointMonitor
|
||||
|
||||
monitor = EndpointMonitor(
|
||||
kind='process', displayname=displayname, processname=processname,
|
||||
source='local', isactive=True,
|
||||
)
|
||||
db.session.add(monitor)
|
||||
db.session.flush()
|
||||
return monitor
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pure computer
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_compute_drift_flags_attached_monitor_not_reported_running(
|
||||
client, db, app):
|
||||
"""An expected-running monitor absent from the snapshot is drift."""
|
||||
from plugins.endpoints.services.reconciler import compute_drift
|
||||
|
||||
asset, ctype = _computer(db, 'DRIFT01', 'CMM PC')
|
||||
monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
drift = compute_drift(asset.assetid, [])
|
||||
assert len(drift) == 1
|
||||
assert drift[0]['name'] == 'dncMain.exe'
|
||||
assert drift[0]['kind'] == 'process'
|
||||
assert drift[0]['expected'] == 'running'
|
||||
assert drift[0]['observed'] == 'notrunning'
|
||||
|
||||
|
||||
def test_compute_drift_empty_when_monitor_reported_running(client, db, app):
|
||||
"""A monitor the snapshot reports running is not drift."""
|
||||
from plugins.endpoints.services.reconciler import compute_drift
|
||||
|
||||
asset, ctype = _computer(db, 'DRIFT02', 'CMM PC')
|
||||
monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
reported = [{'name': 'dncMain.exe', 'kind': 'process', 'state': 'running'}]
|
||||
assert compute_drift(asset.assetid, reported) == []
|
||||
|
||||
|
||||
def test_compute_drift_empty_when_no_monitor_attached(client, db, app):
|
||||
"""No attached monitors means nothing to reconcile."""
|
||||
from plugins.endpoints.services.reconciler import compute_drift
|
||||
|
||||
asset, _ = _computer(db, 'DRIFT03', 'CMM PC')
|
||||
assert compute_drift(asset.assetid, []) == []
|
||||
|
||||
|
||||
def test_monitor_expects_running_is_the_observe_only_seam():
|
||||
"""The 'should be up' decision lives in one function; inactive is excluded."""
|
||||
from plugins.endpoints.services.reconciler import monitor_expects_running
|
||||
|
||||
class _M:
|
||||
isactive = True
|
||||
|
||||
active = _M()
|
||||
inactive = _M()
|
||||
inactive.isactive = False
|
||||
assert monitor_expects_running(active) is True
|
||||
assert monitor_expects_running(inactive) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Check-in write path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_checkin_persists_drift_on_endpoint_state(client, db, app):
|
||||
"""apply_collector_payload caches drift on the state in the same commit."""
|
||||
from plugins.endpoints.plugin import EndpointsPlugin
|
||||
from plugins.endpoints.models import EndpointState
|
||||
|
||||
asset, ctype = _computer(db, 'CHKIN01', 'CMM PC')
|
||||
monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
plugin = EndpointsPlugin()
|
||||
plugin.apply_collector_payload({'hostname': 'CHKIN01', 'services': []})
|
||||
db.session.commit()
|
||||
|
||||
state = EndpointState.query.filter_by(assetid=asset.assetid).first()
|
||||
assert state.driftjson is not None
|
||||
import json
|
||||
drift = json.loads(state.driftjson)
|
||||
assert [d['name'] for d in drift] == ['dncMain.exe']
|
||||
|
||||
|
||||
def test_checkin_clears_drift_when_process_returns(client, db, app):
|
||||
"""A later report that shows the process running clears the cached drift."""
|
||||
from plugins.endpoints.plugin import EndpointsPlugin
|
||||
from plugins.endpoints.models import EndpointState
|
||||
import json
|
||||
|
||||
asset, ctype = _computer(db, 'CHKIN02', 'CMM PC')
|
||||
monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
plugin = EndpointsPlugin()
|
||||
plugin.apply_collector_payload({'hostname': 'CHKIN02', 'services': []})
|
||||
db.session.commit()
|
||||
|
||||
plugin.apply_collector_payload({
|
||||
'hostname': 'CHKIN02',
|
||||
'services': [{'name': 'dncMain.exe', 'kind': 'process',
|
||||
'state': 'running'}],
|
||||
})
|
||||
db.session.commit()
|
||||
|
||||
state = EndpointState.query.filter_by(assetid=asset.assetid).first()
|
||||
assert json.loads(state.driftjson) == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Panel exposure
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_status_panel_surfaces_drift_warning(client, db, app, auth_headers):
|
||||
"""The status panel adds a Needs attention field when drift is cached."""
|
||||
from plugins.endpoints.plugin import EndpointsPlugin
|
||||
|
||||
asset, ctype = _computer(db, 'PANEL01', 'CMM PC')
|
||||
monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
plugin = EndpointsPlugin()
|
||||
plugin.apply_collector_payload({'hostname': 'PANEL01', 'services': []})
|
||||
db.session.commit()
|
||||
|
||||
response = client.get(STATUS_URL.format(asset.assetid), headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
labels = [f['label'] for f in response.get_json()['data']['fields']]
|
||||
assert 'Needs attention' in labels
|
||||
|
||||
|
||||
def test_services_panel_marks_drifted_row(client, db, app, auth_headers):
|
||||
"""The synthetic Unknown row for an expected-running monitor is drift-flagged."""
|
||||
from plugins.endpoints.plugin import EndpointsPlugin
|
||||
|
||||
asset, ctype = _computer(db, 'PANEL02', 'CMM PC')
|
||||
monitor = _process_monitor(db, 'eDNC', 'dncMain.exe')
|
||||
_attach_monitor_to_pctype(db, monitor, ctype.computertypeid)
|
||||
|
||||
plugin = EndpointsPlugin()
|
||||
plugin.apply_collector_payload({'hostname': 'PANEL02', 'services': []})
|
||||
db.session.commit()
|
||||
|
||||
response = client.get(SERVICES_URL.format(asset.assetid), headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
rows = response.get_json()['data']
|
||||
edenc = next((r for r in rows if r['name'] == 'eDNC'), None)
|
||||
assert edenc is not None
|
||||
assert edenc.get('drift') is True
|
||||
141
tests/test_plugins/test_endpoints_shipped_monitor_policy.py
Normal file
141
tests/test_plugins/test_endpoints_shipped_monitor_policy.py
Normal file
@@ -0,0 +1,141 @@
|
||||
"""The shipped shopfloor app monitors ride a read-only Site Policy.
|
||||
|
||||
Migration endpoints0007shippedpolicy offloads the eDNC/UDC/MTConnect monitors
|
||||
from the agent's hardcoded appsettings.json into a product-owned policy bundle:
|
||||
a shipped ConfigurationPolicy "Shopfloor App Monitors" (global scope, one
|
||||
desired-state monitor action per shipped monitor) wrapped in a shipped
|
||||
BusinessPolicy "RPCA Shopfloor App Monitors".
|
||||
|
||||
Run the endpoints chain against a fresh engine, alongside the computers policy
|
||||
tables the bundle writes into, and assert the bundle exists, is source='shipped'
|
||||
(read-only), and attaches every shipped monitor as a desired/running:true action.
|
||||
"""
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import create_engine, inspect, text
|
||||
from alembic.migration import MigrationContext
|
||||
from alembic.operations import Operations
|
||||
|
||||
|
||||
_ROOT = Path(__file__).resolve().parent.parent.parent
|
||||
_ENDPOINTS_VERSIONS = _ROOT / 'plugins' / 'endpoints' / 'migrations' / 'versions'
|
||||
_COMPUTERS_VERSIONS = _ROOT / 'plugins' / 'computers' / 'migrations' / 'versions'
|
||||
|
||||
CONFIG_POLICY_NAME = 'Shopfloor App Monitors'
|
||||
BUSINESS_POLICY_NAME = 'RPCA Shopfloor App Monitors'
|
||||
|
||||
|
||||
def _load(path, prefix):
|
||||
spec = importlib.util.spec_from_file_location(prefix + path.stem, str(path))
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _run_bundle_migrations(connection):
|
||||
"""Build the policy tables (computers 0004), the monitor catalog (endpoints
|
||||
0006) and the shipped bundle (endpoints 0007) on one engine, in order.
|
||||
|
||||
The core baseline creates every plugin's tables before per-plugin chains run,
|
||||
so on a real deploy the computers policy tables already exist when endpoints
|
||||
0007 runs. Here we stand up exactly the three migrations the bundle depends
|
||||
on - the earlier endpoints revisions carry FKs to core tables that a bare
|
||||
engine does not have, and are irrelevant to what 0007 seeds.
|
||||
"""
|
||||
operations = Operations(MigrationContext.configure(connection))
|
||||
operations._install_proxy()
|
||||
try:
|
||||
_load(_COMPUTERS_VERSIONS / '0004_policies.py',
|
||||
'computers_mig_').upgrade()
|
||||
_load(_ENDPOINTS_VERSIONS / '0006_endpoint_monitors.py',
|
||||
'endpoints_mig_0006_').upgrade()
|
||||
_load(_ENDPOINTS_VERSIONS / '0007_shipped_monitor_policy.py',
|
||||
'endpoints_mig_0007_').upgrade()
|
||||
finally:
|
||||
operations._remove_proxy()
|
||||
connection.commit()
|
||||
|
||||
|
||||
def test_shipped_bundle_seeds_read_only_site_policy():
|
||||
engine = create_engine('sqlite://')
|
||||
with engine.connect() as connection:
|
||||
_run_bundle_migrations(connection)
|
||||
|
||||
config = connection.execute(text(
|
||||
'SELECT configurationpolicyid, source, scopejson '
|
||||
'FROM configurationpolicies WHERE name = :name'),
|
||||
{'name': CONFIG_POLICY_NAME}).first()
|
||||
assert config is not None, 'shipped Configuration Policy was not seeded'
|
||||
config_id, config_source, scopejson = config
|
||||
assert config_source == 'shipped'
|
||||
assert json.loads(scopejson) == {'type': 'true'}
|
||||
|
||||
business = connection.execute(text(
|
||||
'SELECT businesspolicyid, source FROM businesspolicies '
|
||||
'WHERE name = :name'), {'name': BUSINESS_POLICY_NAME}).first()
|
||||
assert business is not None, 'shipped Business Policy was not seeded'
|
||||
business_id, business_source = business
|
||||
assert business_source == 'shipped'
|
||||
|
||||
item = connection.execute(text(
|
||||
'SELECT configurationpolicyid FROM businesspolicyitems '
|
||||
'WHERE businesspolicyid = :id'), {'id': business_id}).first()
|
||||
assert item is not None and item[0] == config_id
|
||||
|
||||
|
||||
def test_shipped_bundle_attaches_every_shipped_monitor():
|
||||
engine = create_engine('sqlite://')
|
||||
with engine.connect() as connection:
|
||||
_run_bundle_migrations(connection)
|
||||
|
||||
monitor_ids = {row[0] for row in connection.execute(text(
|
||||
'SELECT endpointmonitorid FROM endpointmonitors '
|
||||
"WHERE source = 'shipped'"))}
|
||||
assert monitor_ids, 'no shipped monitors to attach'
|
||||
|
||||
config_id = connection.execute(text(
|
||||
'SELECT configurationpolicyid FROM configurationpolicies '
|
||||
'WHERE name = :name'), {'name': CONFIG_POLICY_NAME}).scalar()
|
||||
|
||||
actions = connection.execute(text(
|
||||
'SELECT action, mode, desiredstate FROM configurationpolicyactions '
|
||||
'WHERE configurationpolicyid = :id'), {'id': config_id}).all()
|
||||
|
||||
attached_ids = set()
|
||||
for action, mode, desiredstate in actions:
|
||||
assert action.startswith('monitor:')
|
||||
# Gate, not desired: the shipped bundle reports state and exposes
|
||||
# Start/Stop in the UI, but never auto-starts a process. A desired
|
||||
# running state is a deliberate per-monitor opt-in, not shipped.
|
||||
assert mode == 'gate'
|
||||
assert desiredstate is None
|
||||
attached_ids.add(int(action[len('monitor:'):]))
|
||||
|
||||
assert attached_ids == monitor_ids
|
||||
|
||||
|
||||
def test_shipped_bundle_is_idempotent():
|
||||
engine = create_engine('sqlite://')
|
||||
with engine.connect() as connection:
|
||||
_run_bundle_migrations(connection)
|
||||
# Re-run only the shipped-bundle migration; it must not duplicate rows.
|
||||
operations = Operations(MigrationContext.configure(connection))
|
||||
operations._install_proxy()
|
||||
try:
|
||||
_load(_ENDPOINTS_VERSIONS / '0007_shipped_monitor_policy.py',
|
||||
'endpoints_mig_rerun_').upgrade()
|
||||
finally:
|
||||
operations._remove_proxy()
|
||||
connection.commit()
|
||||
|
||||
config_count = connection.execute(text(
|
||||
'SELECT COUNT(*) FROM configurationpolicies WHERE name = :name'),
|
||||
{'name': CONFIG_POLICY_NAME}).scalar()
|
||||
business_count = connection.execute(text(
|
||||
'SELECT COUNT(*) FROM businesspolicies WHERE name = :name'),
|
||||
{'name': BUSINESS_POLICY_NAME}).scalar()
|
||||
assert config_count == 1
|
||||
assert business_count == 1
|
||||
Reference in New Issue
Block a user