cesnimda
237 lines
12 KiB
Python
237 lines
12 KiB
Python
"""Single-user policy tests; no real credentials, models or production services."""
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
from datetime import datetime,timezone,timedelta
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import local_mode as local
|
|
import controller as c
|
|
import test_controller as fixture
|
|
|
|
|
|
class LocalPolicyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='jobjakt-single-user-test-')
|
|
self.base = Path(self.temp.name)
|
|
self.root = self.base / 'control'
|
|
self.repo = self.base / 'worker'
|
|
self.repo.mkdir()
|
|
self.config = dict(mode='single-user-local', executionPolicy='operator-controlled',
|
|
ownerSid='S-1-5-21-100', runnerSid='S-1-5-21-100', repository=str(self.repo),
|
|
workerCodeXHome=str(self.root / 'auth'), normalCodeXHome=str(self.base / 'normal-codex'),
|
|
workerTemp=str(self.base / 'worker-temp'), toolReadRoots=[], trustedCodexOverrides=[])
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def test_same_windows_identity_supported(self):
|
|
local.validate_identity(self.config, self.config['ownerSid'])
|
|
|
|
def test_different_or_missing_sid_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
local.validate_identity(self.config, 'S-1-5-21-999')
|
|
|
|
def test_existing_normal_auth_home_is_rejected(self):
|
|
self.config['workerCodeXHome'] = self.config['normalCodeXHome']
|
|
with self.assertRaisesRegex(ValueError, 'auth'):
|
|
local.validate_config(self.root, self.config)
|
|
|
|
def test_auth_home_inside_worker_is_rejected(self):
|
|
self.config['workerCodeXHome'] = str(self.repo / 'auth')
|
|
with self.assertRaises(ValueError):
|
|
local.validate_config(self.root,self.config)
|
|
|
|
def test_profile_denies_root_control_auth_and_normal_configuration(self):
|
|
rules = local.filesystem_rules(self.root,self.config)
|
|
self.assertEqual('deny',rules[':root'])
|
|
self.assertEqual('read',rules[':minimal'])
|
|
for path in [self.root,Path(self.config['workerCodeXHome']),Path(self.config['normalCodeXHome'])]:
|
|
self.assertEqual('deny',rules[path.resolve().as_posix()])
|
|
self.assertEqual('write',rules[self.repo.resolve().as_posix()])
|
|
self.assertEqual('read',rules[(self.repo / '.git').resolve().as_posix()])
|
|
|
|
def test_mxc_is_explicit_without_legacy_or_unrestricted_fallback(self):
|
|
args = local.permission_args(self.root,self.config,'exec')
|
|
self.assertIn('windows.sandbox="mxc"',args)
|
|
self.assertIn('permissions.jobjakt-local.network.enabled=false',args)
|
|
self.assertNotIn('--sandbox',args)
|
|
self.assertNotIn('--dangerously-bypass-approvals-and-sandbox',args)
|
|
for value in ('features.apps=false','features.plugins=false','features.remote_plugin=false','mcp_servers={}'):
|
|
self.assertIn(value,args)
|
|
|
|
def test_plugins_are_not_an_implicit_same_user_capability(self):
|
|
self.config['trustedCodexOverrides'] = [{'key':'mcp_servers.any.command','toml':'"unsafe"'}]
|
|
with self.assertRaisesRegex(ValueError,'plugin'):
|
|
local.validate_config(self.root,self.config)
|
|
|
|
def test_tool_or_temp_roots_cannot_reopen_protected_parent(self):
|
|
self.config['toolReadRoots']=[str(self.base)]
|
|
with self.assertRaises(ValueError):
|
|
local.validate_config(self.root,self.config)
|
|
self.config['toolReadRoots']=[]
|
|
self.config['workerTemp']=str(self.base)
|
|
with self.assertRaises(ValueError):
|
|
local.validate_config(self.root,self.config)
|
|
|
|
def test_operator_controlled_policy_cannot_launch_a_worker(self):
|
|
with self.assertRaisesRegex(ValueError,'operator-controlled'):
|
|
local.require_unattended(self.root,self.config)
|
|
|
|
def test_missing_or_stale_qualification_never_enables_unattended(self):
|
|
self.config['executionPolicy'] = 'qualified-unattended'
|
|
with self.assertRaises(ValueError):
|
|
local.require_unattended(self.root,self.config)
|
|
|
|
def test_probe_summary_rejects_partial_and_failed_boundaries(self):
|
|
result = dict(positiveControl=True,networkDenied=True,signingKeyDenied=True,
|
|
files=[dict(readDenied=True,writeDenied=False)])
|
|
self.assertFalse(local.probes_passed(result))
|
|
result['files'][0]['writeDenied'] = True
|
|
self.assertFalse(local.probes_passed(result)) # required categories cannot be omitted
|
|
|
|
def test_same_chatgpt_auth_is_explicit_and_tokens_are_not_copied(self):
|
|
normal = Path(self.config['normalCodeXHome'])
|
|
normal.mkdir()
|
|
(normal/'auth.json').write_text('synthetic-existing-token-do-not-copy')
|
|
with mock.patch.object(local,'detect_identity',return_value=('synthetic-user',self.config['ownerSid'])), \
|
|
mock.patch.dict(local.os.environ,{'CODEX_HOME':str(normal)}):
|
|
config = local.configure_current_user(self.root,dict(self.config))
|
|
self.assertEqual(config['ownerSid'],config['runnerSid'])
|
|
self.assertEqual([],list(Path(config['workerCodeXHome']).iterdir()))
|
|
args = local.auth_command({**config,'codexExecutable':'codex'})
|
|
self.assertIn('forced_login_method="chatgpt"',args)
|
|
self.assertIn('cli_auth_credentials_store="file"',args)
|
|
self.assertNotIn('--with-api-key',args)
|
|
self.assertEqual('synthetic-existing-token-do-not-copy',(normal/'auth.json').read_text())
|
|
|
|
def test_failed_probe_and_unconfigured_owner_key_cannot_qualify(self):
|
|
self.root.mkdir()
|
|
self.config['executionPolicy']='qualified-unattended'
|
|
valid = dict(positiveControl=True,networkDenied=True,signingKeyDenied=True,
|
|
files=[dict(category=k,readDenied=True,writeDenied=True) for k in sorted(local.CATEGORIES)])
|
|
report = dict(status='PASS',failure=None,binding=local.binding(self.root,self.config),issuedUtc=datetime.now(timezone.utc).isoformat(),probes=valid,ownerKeyDenied=False)
|
|
c.atomic(self.root/'qualification.json',report)
|
|
with self.assertRaises(ValueError):
|
|
local.require_unattended(self.root,self.config)
|
|
|
|
report['ownerKeyDenied']=True
|
|
report['probes']['networkDenied']=False
|
|
c.atomic(self.root/'qualification.json',report)
|
|
with self.assertRaises(ValueError):
|
|
local.require_unattended(self.root,self.config)
|
|
report['probes']['networkDenied']=True
|
|
c.atomic(self.root/'qualification.json',report)
|
|
local.require_unattended(self.root,self.config)
|
|
report['issuedUtc']=(datetime.now(timezone.utc)-timedelta(days=2)).isoformat()
|
|
c.atomic(self.root/'qualification.json',report)
|
|
with self.assertRaises(ValueError):
|
|
local.require_unattended(self.root,self.config)
|
|
|
|
def test_blocked_status_rejects_otherwise_successful_probes(self):
|
|
self.root.mkdir()
|
|
self.config['executionPolicy']='qualified-unattended'
|
|
c.atomic(self.root/'qualification.json',dict(status='BLOCKED',failure='cleanup failed',
|
|
binding=local.binding(self.root,self.config),issuedUtc=datetime.now(timezone.utc).isoformat(),ownerKeyDenied=True,
|
|
probes=dict(positiveControl=True,networkDenied=True,signingKeyDenied=True,
|
|
files=[dict(category=k,readDenied=True,writeDenied=True) for k in sorted(local.CATEGORIES)])))
|
|
with self.assertRaises(ValueError):
|
|
local.require_unattended(self.root,self.config)
|
|
|
|
def test_noninteractive_approval_is_refused(self):
|
|
c.atomic(self.root/'config.json',self.config)
|
|
c.atomic(self.root/'backlog.json',{})
|
|
with mock.patch.object(c.sys.stdin,'isatty',return_value=False):
|
|
with self.assertRaisesRegex(c.GateError,'interactive terminal'):
|
|
c.main(['approve','--control-root',str(self.root)])
|
|
|
|
|
|
class LocalFlowTests(unittest.TestCase):
|
|
tearDown=fixture.EngineTests.tearDown
|
|
git_at=staticmethod(fixture.EngineTests.git_at)
|
|
run_engine=fixture.EngineTests.run_engine
|
|
|
|
def setUp(self):
|
|
fixture.EngineTests.setUp(self)
|
|
self.config.update(mode='single-user-local',executionPolicy='qualified-unattended',
|
|
ownerSid='S-1-5-21-100',runnerSid='S-1-5-21-100',
|
|
normalCodeXHome=str(self.root/'normal-codex'),workerCodeXHome=str(self.control/'auth'),
|
|
workerTemp=str(self.root/'worker-temp'),toolReadRoots=[],trustedCodexOverrides=[])
|
|
Path(self.config['workerCodeXHome']).mkdir()
|
|
Path(self.config['workerTemp']).mkdir()
|
|
c.atomic(self.control/'qualification.json',dict(status='PASS',failure=None,binding=local.binding(self.control,self.config),
|
|
issuedUtc=datetime.now(timezone.utc).isoformat(),ownerKeyDenied=True,
|
|
probes=dict(positiveControl=True,networkDenied=True,signingKeyDenied=True,
|
|
files=[dict(category=k,readDenied=True,writeDenied=True) for k in sorted(local.CATEGORIES)])))
|
|
|
|
def test_single_user_subprocess_protocol_and_ordered_publication(self):
|
|
self.engine.runner=fixture.ProtocolRunner()
|
|
self.assertEqual('COMPLETED',self.run_engine()['stages']['TWO']['status'])
|
|
|
|
def test_single_user_missing_approval_stops_before_worker(self):
|
|
self.engine.approval_verifier=lambda request:False
|
|
self.assertEqual('AWAITING_APPROVAL',self.run_engine()['stages']['ONE']['status'])
|
|
self.assertEqual(0,self.runner.calls)
|
|
|
|
def test_single_user_pause_and_publication_recovery(self):
|
|
self.engine.pause()
|
|
self.run_engine()
|
|
self.assertEqual(0,self.runner.calls)
|
|
self.engine.resume()
|
|
publish=self.engine.publish
|
|
self.engine.publish=lambda *args: (_ for _ in ()).throw(c.PublicationError('synthetic outage'))
|
|
self.assertEqual('BLOCKED',self.run_engine()['stages']['ONE']['status'])
|
|
self.assertEqual(1,self.runner.calls)
|
|
self.engine.publish=publish
|
|
self.assertEqual('COMPLETED',self.run_engine()['stages']['TWO']['status'])
|
|
self.assertEqual(2,self.runner.calls)
|
|
|
|
def test_single_user_concurrent_controller_and_interrupted_guard(self):
|
|
with c.RepoLock(self.repo):
|
|
with self.assertRaises(c.GateError):
|
|
self.run_engine()
|
|
state=self.engine.store.read()
|
|
state['stages']['ONE']['status']='RUNNING'
|
|
self.engine.store.write(state)
|
|
with self.assertRaises(c.GateError):
|
|
self.run_engine()
|
|
|
|
def test_single_user_main_checkout_is_still_prohibited(self):
|
|
self.git_at(self.repo,'switch','-c','main')
|
|
self.assertEqual('BLOCKED',self.run_engine()['stages']['ONE']['status'])
|
|
self.assertEqual(0,self.runner.calls)
|
|
|
|
def test_pending_recovery_can_be_signed_from_idle_operator_terminal(self):
|
|
self.config['ownerCertificateThumbprint']='synthetic-owner-key'
|
|
c.atomic(self.control/'config.json',self.config)
|
|
c.atomic(self.control/'backlog.json',self.manifest)
|
|
state=self.engine.store.read()
|
|
state['stages']['ONE']['status']='RUNNING'
|
|
self.engine.store.write(state)
|
|
self.engine.approval_verifier=lambda r:False
|
|
def answer(prompt):
|
|
if prompt.startswith('Type '):
|
|
return prompt.split()[1]
|
|
if prompt.startswith('Disposition'):
|
|
return 'Approve'
|
|
return 'Operator inspected the stopped synthetic process tree'
|
|
with mock.patch.object(c,'Engine',return_value=self.engine):
|
|
self.assertEqual(3,c.main(['recover','--control-root',str(self.control)]))
|
|
run=c.subprocess.run
|
|
def operator_command(argv,**kwargs):
|
|
return mock.Mock(returncode=0) if 'Sign' in argv else run(argv,**kwargs)
|
|
with mock.patch.object(c.sys.stdin,'isatty',return_value=True), \
|
|
mock.patch.object(local,'detect_identity',return_value=('synthetic-user',self.config['ownerSid'])), \
|
|
mock.patch('builtins.input',side_effect=answer), \
|
|
mock.patch.object(c.subprocess,'run',side_effect=operator_command) as signed:
|
|
self.assertEqual(0,c.main(['approve','--control-root',str(self.control)]))
|
|
self.assertIn('Sign',signed.call_args.args[0])
|
|
self.assertTrue(self.engine.store.read()['recoveryRequired']) # signing alone never marks recovery complete
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|