5cd5332397
This commit adds: - Database migrations for app_delta fingerprints and client certificates - Scripts for certificate generation and fingerprint backfilling - Comprehensive security audit reports and findings tracking - Implementation documentation for certificate authentication - Test utilities for certificate renewal and fingerprinting - GE-ENFORCE collector fix documentation Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
371 lines
13 KiB
Python
371 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 pytest
|
|
import hashlib
|
|
from datetime import datetime
|
|
|
|
from shopdb.extensions import db
|
|
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()
|
|
|
|
|
|
class TestApplicationDelta:
|
|
"""Test delta-based application inventory updates."""
|
|
|
|
@pytest.fixture
|
|
def plugin(self):
|
|
return EndpointsPlugin()
|
|
|
|
@pytest.fixture
|
|
def asset_id(self, db_session):
|
|
"""Create a test asset."""
|
|
from shopdb.core.models import Asset
|
|
asset = Asset(assettag='TEST-001')
|
|
db_session.add(asset)
|
|
db_session.commit()
|
|
return asset.assetid
|
|
|
|
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_session):
|
|
"""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_session):
|
|
"""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_session):
|
|
"""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_session):
|
|
"""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_session):
|
|
"""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.app_fingerprint is None for app in apps)
|
|
|
|
def test_empty_inventory(self, plugin, asset_id, db_session):
|
|
"""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_session):
|
|
"""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_session):
|
|
"""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_session):
|
|
"""Create a test asset."""
|
|
from shopdb.core.models import Asset
|
|
asset = Asset(assettag='TEST-SAFETY-001')
|
|
db_session.add(asset)
|
|
db_session.commit()
|
|
return asset.assetid
|
|
|
|
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)
|