tegwick
bda9381f07
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
103 lines
4.8 KiB
Python
103 lines
4.8 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
|
|
import pytest
|
|
|
|
from informed_decision.container import prepare_configuration, private_directory, single_writer
|
|
from informed_decision.runtime import Runtime
|
|
from informed_decision.store import Store
|
|
|
|
|
|
@pytest.fixture
|
|
def projected(tmp_path):
|
|
data=tmp_path/'data';data.mkdir()
|
|
run=tmp_path/'run';run.mkdir()
|
|
config={'schema':'informed-decision.review-runtime.v1','evidence_db':str(data/'private/review.sqlite'),
|
|
'approval_origin':'http://approval-engine.approval-engine.svc.cluster.local:8080',
|
|
'policy':{'origin':'http://flex-auth-review.flex-auth.svc.cluster.local:8080',
|
|
'package':'fixture','version':'v1','package_digest':'sha256:'+'a'*64,
|
|
'caller_token_file':str(tmp_path/'not-provisioned-caller')},
|
|
'audit':{'origin':'http://audit-core.audit-core.svc.cluster.local:8080',
|
|
'sender_token_file':str(tmp_path/'not-provisioned-sender')}}
|
|
payload=tmp_path/'projected';payload.write_text(json.dumps(config));payload.chmod(0o444)
|
|
link=tmp_path/'runtime.json';link.symlink_to(payload)
|
|
return link,data,run,config
|
|
|
|
|
|
def test_projected_config_becomes_private_without_reading_or_copying_credentials(projected):
|
|
source,data,run,expected=projected
|
|
result,evidence=prepare_configuration(source,data_root=data,run_root=run)
|
|
assert not result.is_symlink() and stat.S_IMODE(result.stat().st_mode)==0o600
|
|
assert result.stat().st_uid==os.getuid() and json.loads(result.read_text())==expected
|
|
assert stat.S_IMODE(evidence.stat().st_mode)==0o700
|
|
runtime=Runtime.from_file(result)
|
|
assert not runtime.pump.ready()
|
|
assert not Path(expected['policy']['caller_token_file']).exists()
|
|
assert not Path(expected['audit']['sender_token_file']).exists()
|
|
assert sorted(p.name for p in result.parent.iterdir())==['runtime.json']
|
|
|
|
|
|
@pytest.mark.parametrize('fault',['oversize','outside_db','unsafe_directory','symlink_directory'])
|
|
def test_unsafe_container_storage_and_config_refused(projected,tmp_path,fault):
|
|
source,data,run,config=projected
|
|
source.resolve().chmod(0o644)
|
|
if fault=='oversize':source.write_text('x'*16385)
|
|
if fault=='outside_db':
|
|
config['evidence_db']=str(tmp_path/'outside.sqlite');source.write_text(json.dumps(config))
|
|
if fault=='unsafe_directory':(data/'private').mkdir(mode=0o755)
|
|
if fault=='symlink_directory':(data/'private').symlink_to(run)
|
|
with pytest.raises((ValueError,FileExistsError)):
|
|
prepare_configuration(source,data_root=data,run_root=run)
|
|
assert not (data/'private/review.sqlite').exists()
|
|
|
|
|
|
def test_second_service_cannot_open_the_same_evidence_volume(tmp_path):
|
|
private=private_directory(tmp_path/'private')
|
|
with single_writer(private):
|
|
with pytest.raises(ValueError,match='another review service'):
|
|
with single_writer(private):pass
|
|
with single_writer(private):pass # Process/holder release permits restart.
|
|
|
|
|
|
@pytest.mark.parametrize('fault',['symlink','hardlink','mode'])
|
|
def test_unsafe_service_lock_refused(tmp_path,fault):
|
|
private=private_directory(tmp_path/'private');lock=private/'serve.lock'
|
|
if fault=='symlink':lock.symlink_to(tmp_path/'outside')
|
|
else:
|
|
lock.touch(mode=0o600)
|
|
if fault=='hardlink':os.link(lock,private/'other')
|
|
else:lock.chmod(0o644)
|
|
with pytest.raises((ValueError,OSError)):
|
|
with single_writer(private):pass
|
|
|
|
|
|
def test_reconfiguration_preserves_persisted_evidence(projected):
|
|
source,data,run,config=projected
|
|
result,_=prepare_configuration(source,data_root=data,run_root=run)
|
|
first=Runtime.from_file(result);digest=first.controller.store.put_document(b'private fixture content')
|
|
source.resolve().chmod(0o644)
|
|
config['policy']['version']='v2';source.write_text(json.dumps(config))
|
|
result,_=prepare_configuration(source,data_root=data,run_root=run)
|
|
second=Runtime.from_file(result)
|
|
assert second.controller.policy.version=='v2'
|
|
with second.controller.store._connection() as db:
|
|
assert db.execute('SELECT content FROM documents WHERE digest=?',(digest,)).fetchone()[0]==b'private fixture content'
|
|
|
|
|
|
@pytest.mark.parametrize('host',[None,'0.0.0.0','localhost'])
|
|
def test_listener_exposure_requires_explicit_supported_setting(monkeypatch,host):
|
|
import waitress
|
|
from informed_decision import web
|
|
monkeypatch.delenv('INFD_REVIEW_CONFIG',raising=False)
|
|
monkeypatch.setenv('INFD_KEYCAPE_ISSUER','https://keycape.test')
|
|
monkeypatch.delenv('INFD_LISTEN_HOST',raising=False)
|
|
if host:monkeypatch.setenv('INFD_LISTEN_HOST',host)
|
|
observed=[]
|
|
monkeypatch.setattr(waitress,'serve',lambda app,**kwargs:observed.append(kwargs))
|
|
if host=='localhost':
|
|
with pytest.raises(ValueError):web.main()
|
|
assert not observed
|
|
else:
|
|
web.main();assert observed[0]['host']==(host or '127.0.0.1')
|