admin_rb
132 lines
6.8 KiB
Python
132 lines
6.8 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
import tomllib
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0,str(ROOT/'deploy'))
|
|
spec=importlib.util.spec_from_file_location('deployment_v2',ROOT/'deploy/deploy.py')
|
|
deployment=importlib.util.module_from_spec(spec);sys.modules[spec.name]=deployment;spec.loader.exec_module(deployment)
|
|
|
|
def test_units_have_no_sudo_or_root_runtime():
|
|
for mode,executor,user in [('serve',False,'aim-web'),('worker',False,'aim-web'),('executor',True,'svc_bf-ansible')]:
|
|
text=deployment.unit_text(user,991,Path('/etc/ansible/scripts/config/webgui.toml'),mode,executor=executor,executor_group=1001 if executor else None,supplementary_group=991 if executor else None,executor_local_home='/home/svc_bf-ansible' if executor else None)
|
|
assert f'User={user}\n'in text
|
|
assert 'NoNewPrivileges=true' in text
|
|
assert 'CapabilityBoundingSet=\n'in text
|
|
assert 'CAP_SETUID'not in text and 'sudo'not in text
|
|
assert 'RuntimeDirectory=aim-web-executor' in text if executor else 'RuntimeDirectory='not in text
|
|
if executor: assert 'RuntimeDirectoryMode=0711' in text
|
|
if executor:
|
|
assert 'Group=1001\n' in text
|
|
assert 'SupplementaryGroups=991\n' in text
|
|
assert 'Environment=HOME=/var/lib/aim-web-executor' in text
|
|
assert 'ReadWritePaths=/var/lib/aim-web-executor /var/lib/aim-web-executor/.ansible/tmp /home/svc_bf-ansible/.ansible/tmp -/etc/ansible/inventories' in text
|
|
assert 'core-staging-check' in text
|
|
|
|
|
|
def test_no_private_core_import_and_no_old_stage_bridge():
|
|
text=(ROOT/'deploy/deploy.py').read_text()
|
|
assert 'from aim.config'not in text and 'sys.path.insert'not in text
|
|
assert 'key_export.py'not in text
|
|
assert 'core_transport="stdio"'in text
|
|
assert '--migrate-core'in text
|
|
assert "'db','migrate'"in text
|
|
assert text.index("os.replace(stage, target)")<text.index('atomic_bytes(EXECUTOR_UNIT')
|
|
|
|
|
|
def test_site_profile_and_versions():
|
|
from aim_webgui.config import Settings
|
|
from aim_webgui import __version__,SCHEMA_VERSION,HTTP_API_VERSION
|
|
settings=Settings.load(ROOT/'deploy/webgui.example.toml')
|
|
assert settings.core_transport=='unix' and settings.core_executor_user=='svc_bf-ansible'
|
|
assert settings.host=='127.0.0.1' and settings.execution_require_approval is False
|
|
assert len(settings.execution_playbooks)==14
|
|
assert __version__=='2.1.0rc9' and SCHEMA_VERSION==5 and HTTP_API_VERSION==2
|
|
metadata=tomllib.loads((ROOT/'pyproject.toml').read_text())
|
|
assert metadata['tool']['aim-web']['http-api']==2
|
|
|
|
|
|
def test_stage_and_rollback_preserve_modes(tmp_path):
|
|
target=tmp_path/'file';deployment.atomic_bytes(target,b'first',0o640)
|
|
deployment.atomic_bytes(target,b'second',0o600)
|
|
assert target.read_bytes()==b'second' and target.stat().st_mode&0o777==0o600
|
|
link=tmp_path/'link';link.symlink_to(target)
|
|
import pytest
|
|
with pytest.raises(ValueError):deployment.atomic_bytes(link,b'unsafe')
|
|
assert target.read_bytes()==b'second'
|
|
|
|
|
|
def test_release_manifest_detects_changed_or_traversing_payload(tmp_path):
|
|
import hashlib
|
|
import pytest
|
|
files={'pyproject.toml':b'project','deploy/deploy.py':b'installer','src/aim_webgui/__init__.py':b'version'}
|
|
for name,data in files.items():
|
|
p=tmp_path/name;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(data)
|
|
manifest=tmp_path/'MANIFEST.sha256'
|
|
original=''.join(hashlib.sha256(data).hexdigest()+' '+name+'\n'for name,data in files.items())
|
|
manifest.write_text(original);deployment.verify_release(tmp_path)
|
|
dotted=''.join(hashlib.sha256(data).hexdigest()+' ./'+name+'\n' for name,data in files.items())
|
|
manifest.write_text(dotted);deployment.verify_release(tmp_path)
|
|
manifest.write_text(original)
|
|
(tmp_path/'pyproject.toml').write_bytes(b'changed')
|
|
with pytest.raises(ValueError,match='integrity'):deployment.verify_release(tmp_path)
|
|
manifest.write_text('0'*64+' ../outside\n')
|
|
with pytest.raises(ValueError,match='manifest'):deployment.verify_release(tmp_path)
|
|
|
|
|
|
def test_home_and_direct_profile_defaults():
|
|
from aim_webgui.config import Settings
|
|
from dataclasses import replace
|
|
import pytest
|
|
settings=Settings.load(ROOT/'deploy/webgui.example.toml')
|
|
assert settings.core_home==Path('/var/lib/aim-web-executor')
|
|
with pytest.raises(ValueError):replace(settings,core_home=settings.state_dir).validate()
|
|
browsing=Settings.load(ROOT/'deploy/profiles/direct-npm.toml')
|
|
assert not browsing.execution_enabled and not browsing.credentials_enabled
|
|
|
|
|
|
def test_executor_identity_preserves_primary_gid_and_scopes_web_group():
|
|
text=(ROOT/'deploy/deploy.py').read_text()
|
|
assert 'os.setgid(account.pw_gid)' in text
|
|
assert "os.getgrouplist(user,account.pw_gid) + [web.pw_gid]" in text
|
|
executor=(ROOT/'src/aim_webgui/core/executor.py').read_text()
|
|
assert 'os.chown(parent,-1,client_gid)' not in executor
|
|
assert "stat.S_IMODE(st.st_mode)!=0o711" in executor
|
|
assert 'os.chown(path,-1,client_gid)' in executor
|
|
|
|
|
|
def test_executor_socket_readiness_waits_for_type_simple_bind(monkeypatch,tmp_path):
|
|
attempts=[]
|
|
def fake_validate(*args,**kwargs):
|
|
attempts.append(1)
|
|
if len(attempts)<3:
|
|
raise ValueError('Managed executor socket is missing or a symlink')
|
|
class Result:
|
|
returncode=0
|
|
monkeypatch.setattr(deployment,'validate_executor_runtime_permissions',fake_validate)
|
|
monkeypatch.setattr(deployment.subprocess,'run',lambda *a,**k: Result())
|
|
monkeypatch.setattr(deployment.time,'sleep',lambda *_: None)
|
|
deployment.wait_executor_runtime_permissions('aim-web','svc-bf',tmp_path/'core.sock',timeout=1)
|
|
assert len(attempts)==3
|
|
|
|
|
|
def test_executor_socket_readiness_fails_if_service_exits(monkeypatch,tmp_path):
|
|
import pytest
|
|
monkeypatch.setattr(deployment,'validate_executor_runtime_permissions',lambda *a,**k: (_ for _ in ()).throw(ValueError('missing socket')))
|
|
class Result:
|
|
returncode=3
|
|
monkeypatch.setattr(deployment.subprocess,'run',lambda *a,**k: Result())
|
|
with pytest.raises(ValueError,match='exited before'):
|
|
deployment.wait_executor_runtime_permissions('aim-web','svc-bf',tmp_path/'core.sock',timeout=1)
|
|
|
|
def test_restore_guard_probes_restored_adapter_as_executor_before_start(monkeypatch,tmp_path):
|
|
d=deployment.Deployment(tmp_path,'root',0);calls=[]
|
|
monkeypatch.setattr(d,'as_executor',lambda user,cmd,**kw:calls.append((user,cmd)))
|
|
assert d.restored_core_compatible({'venv':str(tmp_path/'old'),'executor_user':'nobody'})
|
|
assert calls[0][0]=='nobody' and str(tmp_path/'old/bin/python')==str(calls[0][1][0])
|
|
assert 'core_transport="stdio"'in calls[0][1][3]
|
|
def fail(*args,**kwargs):raise deployment.subprocess.CalledProcessError(1,['fixture'])
|
|
monkeypatch.setattr(d,'as_executor',fail)
|
|
assert not d.restored_core_compatible({'venv':str(tmp_path/'old')})
|