f0377726a1
Per-PC credentials (replaces the 09-14 client-certificate attempt): - POST /api/endpoints/enroll issues a credential to a PC that matches exactly one computer by hostname and BIOS serial and has none yet. Anything else (unknown or duplicate hostname, serial mismatch, already enrolled) waits as a request an admin approves or rejects. The agent's persisted nonce binds an approved request to the PC that asked. - One auth helper serves every agent route. With X-Endpoint-Credential the asset comes from the credential and a request naming another PC gets 403 ASSET_MISMATCH. The shared key still works for PCs that have not enrolled, is refused (401 SHARED_KEY_REFUSED) for any PC that has, and can be switched off. A revoked credential gets 401 CREDENTIAL_INVALID so the agent re-enrolls. - The core collector accepts the credential for the computers, printers and endpoints collectors through two new plugin hooks exposed on shopdb.api (contract 0.33.0), and refuses the shared key for enrolled PCs. - Admin API: enrollment list, approve, reject, revoke (un-enroll), settings. Migration API: per-PC readiness with blockers, and upgrade waves that package the agent once and queue the existing agent upgrade per PC. - Migration 0023 creates endpointcredentials, endpointenrollrequests and endpointmigrationwaves, drops the certificate columns from assets where test databases have them, and renames app_fingerprint to appfingerprint. 0021 and 0022 keep their ids but no longer alter the core assets table. - Removed: client_cert.py, certificate renewal, the unauthenticated /test/client-cert route, nginx client certificate forwarding and the agent certificate script. The client-cert header could be forged on IIS, the key was readable by users, and the shared key remained a fallback. - GET /settings/server-cert returns the SPKI SHA-256 pin for GE-Enforce. Review fixes: - report_command_result used g without importing it; every result was a 500. - collect_log refuses setup answer files, appsettings, agent directories and key files at enqueue. - Fleet alerts: alerttimeout and alertfleet agree with the migration, the scope is validated and a malformed scope no longer reaches the whole fleet, and the ack race no longer raises NameError. - A refused break-glass override leaves no hold and no audit row. - Agent upgrade checks locks and permissions before repackaging, and writes the package atomically. - Server certificate upload validates the pair before replacing files; generation no longer builds an openssl config from user input; audit rows are actually written. - Health evaluation runs in a savepoint so a caught error cannot lose the check-in. - Required-app policies are found by appid, survive renames, and deleting an app deactivates policies that still name it. The UI and API now say such a policy only takes effect inside a business policy. - Backfill handles duplicate fingerprints and rolls back a failed batch. - Plugin settings routes skip a plugin whose hook raises. - The setup token is no longer printed to the log. - The naming gate checks top-level files again and handles odd file names. - Removed root scratch scripts, one of which carried a token. - Rewrote the fleet alert scheduling and app delta tests on real fixtures.
378 lines
13 KiB
Python
378 lines
13 KiB
Python
"""Tests for application inventory delta-based updates.
|
|
|
|
Tests the _replace_applications() method with fingerprint-based delta logic,
|
|
including safety features: fingerprint validation, delete count verification,
|
|
memory limits, and concurrent update protection.
|
|
"""
|
|
|
|
import hashlib
|
|
|
|
import pytest
|
|
|
|
from plugins.endpoints.models import EndpointApplication
|
|
from plugins.endpoints.plugin import EndpointsPlugin
|
|
|
|
|
|
def compute_test_fingerprint(name, version, publisher):
|
|
"""Compute fingerprint matching agent algorithm."""
|
|
name = name or ''
|
|
version = version or ''
|
|
publisher = publisher or ''
|
|
data = f"{name}|{version}|{publisher}"
|
|
return hashlib.sha256(data.encode('utf-8')).hexdigest()
|
|
|
|
|
|
def _make_asset(db, assetnumber):
|
|
"""A bare computer-type Asset; _replace_applications needs only the id."""
|
|
from shopdb.core.models import Asset, AssetType
|
|
|
|
assettype = AssetType.query.filter_by(assettype='computer').first()
|
|
if assettype is None:
|
|
assettype = AssetType(assettype='computer')
|
|
db.session.add(assettype)
|
|
db.session.flush()
|
|
asset = Asset(assetnumber=assetnumber, name=assetnumber,
|
|
assettypeid=assettype.assettypeid, isactive=True)
|
|
db.session.add(asset)
|
|
db.session.commit()
|
|
return asset.assetid
|
|
|
|
|
|
class TestApplicationDelta:
|
|
"""Test delta-based application inventory updates."""
|
|
|
|
@pytest.fixture
|
|
def plugin(self):
|
|
return EndpointsPlugin()
|
|
|
|
@pytest.fixture
|
|
def asset_id(self, db):
|
|
"""Create a test asset."""
|
|
return _make_asset(db, 'TEST-001')
|
|
|
|
def test_fingerprint_validation_valid(self, plugin):
|
|
"""Valid fingerprints (64 lowercase hex chars) pass validation."""
|
|
valid_fps = [
|
|
'a' * 64, # All lowercase a
|
|
'0' * 64, # All zeros
|
|
'abc123' * 10 + 'abcd', # Mix of hex chars (64 total)
|
|
]
|
|
|
|
for fp in valid_fps:
|
|
assert plugin._is_valid_fingerprint(fp), f"Should accept: {fp}"
|
|
|
|
def test_fingerprint_validation_invalid(self, plugin):
|
|
"""Invalid fingerprints are rejected."""
|
|
invalid_fps = [
|
|
'', # Empty
|
|
'abc', # Too short
|
|
'A' * 64, # Uppercase
|
|
'g' * 64, # Non-hex character
|
|
'abc-def' * 9, # Contains hyphens
|
|
'0' * 63, # One char too short
|
|
'0' * 65, # One char too long
|
|
]
|
|
|
|
for fp in invalid_fps:
|
|
assert not plugin._is_valid_fingerprint(fp), f"Should reject: {fp}"
|
|
|
|
def test_delta_no_changes(self, plugin, asset_id, db):
|
|
"""No apps added/removed when inventory unchanged."""
|
|
# Initial insert: 3 apps
|
|
apps_v1 = [
|
|
{
|
|
'displayname': 'Chrome',
|
|
'displayversion': '117.0',
|
|
'publisher': 'Google',
|
|
'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google')
|
|
},
|
|
{
|
|
'displayname': 'Firefox',
|
|
'displayversion': '119.0',
|
|
'publisher': 'Mozilla',
|
|
'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla')
|
|
},
|
|
{
|
|
'displayname': 'Office',
|
|
'displayversion': '16.0',
|
|
'publisher': 'Microsoft',
|
|
'fingerprint': compute_test_fingerprint('Office', '16.0', 'Microsoft')
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, apps_v1, warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 3
|
|
assert len(warnings) == 0
|
|
assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 3
|
|
|
|
# Second report: same apps (no changes)
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, apps_v1, warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 0 # 0 added
|
|
assert len(warnings) == 0
|
|
assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 3 # Still 3
|
|
|
|
def test_delta_add_remove(self, plugin, asset_id, db):
|
|
"""Detect adds and removes correctly."""
|
|
# Initial: 2 apps
|
|
apps_v1 = [
|
|
{
|
|
'displayname': 'Chrome',
|
|
'displayversion': '117.0',
|
|
'publisher': 'Google',
|
|
'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google')
|
|
},
|
|
{
|
|
'displayname': 'Firefox',
|
|
'displayversion': '119.0',
|
|
'publisher': 'Mozilla',
|
|
'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla')
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
plugin._replace_applications(asset_id, apps_v1, warnings)
|
|
db.session.commit()
|
|
|
|
assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 2
|
|
|
|
# Update: Remove Firefox, Add Office
|
|
apps_v2 = [
|
|
{
|
|
'displayname': 'Chrome',
|
|
'displayversion': '117.0',
|
|
'publisher': 'Google',
|
|
'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google')
|
|
},
|
|
{
|
|
'displayname': 'Office',
|
|
'displayversion': '16.0',
|
|
'publisher': 'Microsoft',
|
|
'fingerprint': compute_test_fingerprint('Office', '16.0', 'Microsoft')
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, apps_v2, warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 1 # 1 added (Office)
|
|
assert len(warnings) == 0
|
|
|
|
apps = EndpointApplication.query.filter_by(assetid=asset_id).all()
|
|
assert len(apps) == 2
|
|
app_names = {app.displayname for app in apps}
|
|
assert app_names == {'Chrome', 'Office'}
|
|
|
|
def test_delta_version_change(self, plugin, asset_id, db):
|
|
"""Version change detected (different fingerprint)."""
|
|
# Initial: Chrome 117.0
|
|
apps_v1 = [
|
|
{
|
|
'displayname': 'Chrome',
|
|
'displayversion': '117.0',
|
|
'publisher': 'Google',
|
|
'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google')
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
plugin._replace_applications(asset_id, apps_v1, warnings)
|
|
db.session.commit()
|
|
|
|
# Update: Chrome 118.0 (different fingerprint)
|
|
apps_v2 = [
|
|
{
|
|
'displayname': 'Chrome',
|
|
'displayversion': '118.0',
|
|
'publisher': 'Google',
|
|
'fingerprint': compute_test_fingerprint('Chrome', '118.0', 'Google')
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, apps_v2, warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 1 # 1 added (new version)
|
|
apps = EndpointApplication.query.filter_by(assetid=asset_id).all()
|
|
assert len(apps) == 1
|
|
assert apps[0].displayversion == '118.0'
|
|
|
|
def test_invalid_fingerprint_skipped(self, plugin, asset_id, db):
|
|
"""Apps with invalid fingerprints are skipped with warning."""
|
|
apps = [
|
|
{
|
|
'displayname': 'Chrome',
|
|
'displayversion': '117.0',
|
|
'publisher': 'Google',
|
|
'fingerprint': 'INVALID' # Too short
|
|
},
|
|
{
|
|
'displayname': 'Firefox',
|
|
'displayversion': '119.0',
|
|
'publisher': 'Mozilla',
|
|
'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla')
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, apps, warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 1 # Only Firefox added
|
|
assert len(warnings) == 1
|
|
assert 'invalid fingerprint' in warnings[0].lower()
|
|
assert 'Chrome' in warnings[0]
|
|
|
|
apps = EndpointApplication.query.filter_by(assetid=asset_id).all()
|
|
assert len(apps) == 1
|
|
assert apps[0].displayname == 'Firefox'
|
|
|
|
def test_legacy_fallback_no_fingerprints(self, plugin, asset_id, db):
|
|
"""Legacy mode works when agent doesn't send fingerprints."""
|
|
apps = [
|
|
{
|
|
'displayname': 'Chrome',
|
|
'displayversion': '117.0',
|
|
'publisher': 'Google',
|
|
# No fingerprint key
|
|
},
|
|
{
|
|
'displayname': 'Firefox',
|
|
'displayversion': '119.0',
|
|
'publisher': 'Mozilla',
|
|
# No fingerprint key
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, apps, warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 2
|
|
assert len(warnings) == 0
|
|
|
|
apps = EndpointApplication.query.filter_by(assetid=asset_id).all()
|
|
assert len(apps) == 2
|
|
assert all(app.appfingerprint is None for app in apps)
|
|
|
|
def test_empty_inventory(self, plugin, asset_id, db):
|
|
"""Empty inventory removes all apps."""
|
|
# Initial: 2 apps
|
|
apps_v1 = [
|
|
{
|
|
'displayname': 'Chrome',
|
|
'displayversion': '117.0',
|
|
'publisher': 'Google',
|
|
'fingerprint': compute_test_fingerprint('Chrome', '117.0', 'Google')
|
|
},
|
|
{
|
|
'displayname': 'Firefox',
|
|
'displayversion': '119.0',
|
|
'publisher': 'Mozilla',
|
|
'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla')
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
plugin._replace_applications(asset_id, apps_v1, warnings)
|
|
db.session.commit()
|
|
|
|
assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 2
|
|
|
|
# Update: Empty list
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, [], warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 0
|
|
assert EndpointApplication.query.filter_by(assetid=asset_id).count() == 0
|
|
|
|
def test_missing_displayname_skipped(self, plugin, asset_id, db):
|
|
"""Apps without displayname are skipped."""
|
|
apps = [
|
|
{
|
|
# No displayname
|
|
'displayversion': '117.0',
|
|
'publisher': 'Google',
|
|
'fingerprint': compute_test_fingerprint('', '117.0', 'Google')
|
|
},
|
|
{
|
|
'displayname': 'Firefox',
|
|
'displayversion': '119.0',
|
|
'publisher': 'Mozilla',
|
|
'fingerprint': compute_test_fingerprint('Firefox', '119.0', 'Mozilla')
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, apps, warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 1 # Only Firefox added
|
|
assert len(warnings) == 1
|
|
assert 'no displayname' in warnings[0].lower()
|
|
|
|
def test_string_truncation(self, plugin, asset_id, db):
|
|
"""Long strings are truncated to column limits."""
|
|
apps = [
|
|
{
|
|
'displayname': 'A' * 300, # Exceeds 255 char limit
|
|
'displayversion': 'B' * 200, # Exceeds 128 char limit
|
|
'publisher': 'C' * 300, # Exceeds 255 char limit
|
|
'fingerprint': compute_test_fingerprint('A' * 300, 'B' * 200, 'C' * 300)
|
|
},
|
|
]
|
|
|
|
warnings = []
|
|
result = plugin._replace_applications(asset_id, apps, warnings)
|
|
db.session.commit()
|
|
|
|
assert result == 1
|
|
app = EndpointApplication.query.filter_by(assetid=asset_id).first()
|
|
assert len(app.displayname) == 255
|
|
assert len(app.displayversion) == 128
|
|
assert len(app.publisher) == 255
|
|
|
|
|
|
class TestApplicationDeltaSafety:
|
|
"""Test safety features: locking, delete verification, memory limits."""
|
|
|
|
@pytest.fixture
|
|
def plugin(self):
|
|
return EndpointsPlugin()
|
|
|
|
@pytest.fixture
|
|
def asset_id(self, db):
|
|
"""Create a test asset."""
|
|
return _make_asset(db, 'TEST-SAFETY-001')
|
|
|
|
def test_compute_fingerprint_consistency(self):
|
|
"""Fingerprint algorithm is deterministic."""
|
|
fp1 = compute_test_fingerprint('Chrome', '117.0', 'Google')
|
|
fp2 = compute_test_fingerprint('Chrome', '117.0', 'Google')
|
|
assert fp1 == fp2
|
|
|
|
# Different inputs produce different fingerprints
|
|
fp3 = compute_test_fingerprint('Chrome', '118.0', 'Google')
|
|
assert fp1 != fp3
|
|
|
|
def test_fingerprint_null_handling(self):
|
|
"""Null values handled correctly (converted to empty string)."""
|
|
fp1 = compute_test_fingerprint('App', None, None)
|
|
fp2 = compute_test_fingerprint('App', '', '')
|
|
assert fp1 == fp2 # None and '' should produce same fingerprint
|
|
|
|
def test_fingerprint_format(self):
|
|
"""Fingerprints are 64 lowercase hex chars."""
|
|
fp = compute_test_fingerprint('Test', '1.0', 'Vendor')
|
|
assert len(fp) == 64
|
|
assert fp.islower()
|
|
assert all(c in '0123456789abcdef' for c in fp)
|