File size: 6,003 Bytes
4d2ab8d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | """Security and compatibility tests for evolution, updates, and local packages."""
import asyncio
import hashlib
from pathlib import Path
import pytest
import main
from agent.evolution.contribution import ContributionQueue
from agent.evolution.sanitizer import sanitize
from agent.evolution.updater import UpdateComponent, UpdateManifest, VerifiedUpdater
from agent.memory.store import Store
from agent.models.manager import LocalIntelligenceManager
from agent.skills.registry import Skill, SkillRegistry
def _file_url(path: Path) -> str:
return path.resolve().as_uri()
def test_collective_learning_is_disabled_and_requires_consent(tmp_path, monkeypatch):
queue = ContributionQueue(Store(tmp_path / 'agent.db'))
monkeypatch.setattr('agent.evolution.contribution.settings.collective_learning_enabled', False)
result = queue.enqueue('routing', {'task_class': 'coding', 'route': 'local'})
assert result == {'accepted': False, 'reason': 'collective_learning_disabled'}
assert queue.pending() == []
def test_sanitizer_rejects_secrets_and_unknown_fields():
secret = 'sk-' + ('synthetic' * 3)
assert not sanitize('knowledge', {'summary': secret}).accepted
assert not sanitize('routing', {'task_class': 'coding', 'unexpected': 'value'}).accepted
def test_untrusted_skill_is_registered_but_never_loaded(tmp_path):
registry = SkillRegistry(tmp_path / 'skills.db')
registry.register(Skill(
skill_id='remote-candidate',
name='Remote candidate',
description='coding helper',
capabilities=('coding',),
instructions='Run an unverified process.',
source='remote',
trust_state='untrusted',
))
context, used = registry.context_for('coding helper', capabilities={'coding'})
assert context == ''
assert used == []
def test_update_component_rejects_traversal_and_insecure_url():
base = {
'id': 'core', 'version': '1.0.0', 'sha256': 'a' * 64,
'destination': 'agent/core.py',
}
with pytest.raises(ValueError, match='HTTPS'):
UpdateComponent.from_dict({**base, 'url': 'http://example.test/core.py'})
with pytest.raises(ValueError, match='traversal'):
UpdateComponent.from_dict({**base, 'url': 'https://example.test/core.py', 'destination': '../core.py'})
def test_verified_update_stages_hash_and_rolls_back_activation(tmp_path, monkeypatch):
root = tmp_path / 'install'
backup = tmp_path / 'backups'
source = tmp_path / 'component.bin'
source.write_bytes(b'verified update')
digest = hashlib.sha256(source.read_bytes()).hexdigest()
component = UpdateComponent.from_dict({
'id': 'core', 'version': '2.0.0', 'url': _file_url(source),
'sha256': digest, 'destination': 'agent/core.bin',
'size_bytes': source.stat().st_size,
})
updater = VerifiedUpdater(root, backup)
staged = asyncio.run(updater.stage([component], tmp_path / 'staging'))
result = updater.activate(staged, '2.0.0')
assert result['activated'] == 1
assert (root / 'agent' / 'core.bin').read_bytes() == b'verified update'
bad_source = tmp_path / 'bad.bin'
bad_source.write_bytes(b'bad')
bad = UpdateComponent.from_dict({
'id': 'bad', 'version': '2.0.0', 'url': _file_url(bad_source),
'sha256': '0' * 64, 'destination': 'agent/bad.bin',
})
with pytest.raises(ValueError, match='hash mismatch'):
asyncio.run(updater.stage([bad], tmp_path / 'bad-staging'))
assert not (root / 'agent' / 'bad.bin').exists()
def test_update_manifest_plan_is_incremental(tmp_path):
component = UpdateComponent(
'core', '2.0.0', (tmp_path / 'source').as_uri(), 'a' * 64, 'agent/core.py'
)
manifest = UpdateManifest('2.0.0', '0.2.0', (component,))
updater = VerifiedUpdater(tmp_path / 'root', tmp_path / 'backup')
assert updater.plan(manifest, {'core': '2.0.0'}) == []
assert updater.plan(manifest, {'core': '1.0.0'}) == [component]
def test_local_package_uninstall_is_dry_run_then_approval(tmp_path, monkeypatch):
root = tmp_path / 'repo'
package = root / 'data' / 'models' / 'managed-package'
package.mkdir(parents=True)
(package / 'manifest.json').write_text('{}', encoding='utf-8')
external = tmp_path / 'external.gguf'
external.write_bytes(b'external')
monkeypatch.setattr('agent.models.manager.ROOT', root)
monkeypatch.setattr('agent.models.manager.settings.local_model_package_path', str(package))
manager = LocalIntelligenceManager()
preview = manager.uninstall_package()
assert preview['executed'] is False
assert package.exists()
removed = manager.uninstall_package(approved=True)
assert removed['executed'] is True
assert not package.exists()
assert external.exists()
def test_local_package_uninstall_rejects_external_directory(tmp_path, monkeypatch):
root = tmp_path / 'repo'
external = tmp_path / 'external-package'
external.mkdir()
monkeypatch.setattr('agent.models.manager.ROOT', root)
monkeypatch.setattr('agent.models.manager.settings.local_model_package_path', str(external))
with pytest.raises(ValueError, match='outside'):
LocalIntelligenceManager().uninstall_package(approved=True)
def test_evolution_status_api_does_not_expose_skill_instructions(tmp_path, monkeypatch):
registry = SkillRegistry(tmp_path / 'skills.db')
registry.register(Skill(
skill_id='safe-skill', name='Safe skill', instructions='private implementation detail'
))
monkeypatch.setattr(main, 'skill_registry', registry)
result = main.evolution_status()
assert result['updates']['verification'] == 'sha256_required'
assert result['collective_learning']['silent_uploads'] is False
assert result['skills'][0]['skill_id'] == 'safe-skill'
assert 'instructions' not in result['skills'][0]
|