404 lines
26 KiB
Python
404 lines
26 KiB
Python
#!/usr/bin/python3
|
|
"""Actual normal HTTP/PTY, renewal and D09 owner acceptance in the private realm.
|
|
|
|
No replacement service, fake APP_READY, surrogate terminal, payment or private
|
|
owner wire injection. Lifecycle requests originate through the native FD10 RPC.
|
|
"""
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import http.cookies
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import runpy
|
|
import secrets
|
|
import shutil
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
PACKAGE=Path('/opt/c01-fixture');EVIDENCE=Path('/var/lib/c01-normal')
|
|
scope=runpy.run_path(str(PACKAGE/'tests/regression/c01-compose.py'))
|
|
provision=scope['d09_module']();owner=provision.owner
|
|
import d09_store as store
|
|
import d09_wire as wire
|
|
generation,policy=scope['generation'],scope['policy']
|
|
sha=scope['sha'];dump=scope['dump'];run=scope['run']
|
|
|
|
|
|
def ledger():return json.loads((owner.ROOT/'ledger.json').read_text())
|
|
def binding():return ledger().get('active_binding') or json.loads(owner.BINDING.read_text())
|
|
def wait(body,label,seconds=180):
|
|
end=time.monotonic()+seconds
|
|
while time.monotonic()<end:
|
|
value=body()
|
|
if value:return value
|
|
time.sleep(.2)
|
|
raise AssertionError('actual acceptance timeout: '+label)
|
|
|
|
|
|
class NativeHTTP:
|
|
def __init__(self):self.cookies={};self.sequence=0
|
|
def request(self,path,value=None):
|
|
headers={'Origin':'http://127.0.0.1:5678','Content-Type':'application/json'}
|
|
if self.cookies.get('csrf_token'):headers['X-CSRF-Token']=self.cookies['csrf_token']
|
|
if self.cookies:headers['Cookie']='; '.join(k+'='+v for k,v in self.cookies.items())
|
|
request=urllib.request.Request('http://127.0.0.1:5678'+path,
|
|
data=json.dumps(value).encode() if value is not None else None,headers=headers)
|
|
with urllib.request.urlopen(request,timeout=40) as response:
|
|
for raw in response.headers.get_all('Set-Cookie',[]):
|
|
values=http.cookies.SimpleCookie();values.load(raw)
|
|
self.cookies.update({k:v.value for k,v in values.items()})
|
|
return json.loads(response.read(1024*1024))
|
|
def rpc(self,method,params=None):
|
|
self.sequence+=1
|
|
result=self.request('/rpc/v1',{'jsonrpc':'2.0','id':self.sequence,'method':method,'params':params})
|
|
if result.get('error') is not None:raise RuntimeError(method+': '+json.dumps(result['error']))
|
|
return result['result']
|
|
def login(self):
|
|
self.cookies={}
|
|
password=json.loads((PACKAGE/'baseline-auth.json').read_text())['password']
|
|
self.rpc('auth.login',{'password':password})
|
|
assert self.cookies,'real authenticated session cookie absent'
|
|
|
|
|
|
def terminal(http):
|
|
record=http.request('/api/terminal/sessions',{'name':'D09 native acceptance'})
|
|
assert record['workspace']=='/home/archipelago/Work' and record['state']=='running'
|
|
nonce=secrets.token_hex(12);marker='C01_PTY_'+nonce
|
|
channel=socket.create_connection(('127.0.0.1',5678),timeout=10)
|
|
key=base64.b64encode(secrets.token_bytes(16)).decode()
|
|
cookie='; '.join(k+'='+v for k,v in http.cookies.items())
|
|
channel.sendall(('GET /ws/terminal?session='+record['id']+' HTTP/1.1\r\nHost: 127.0.0.1:5678\r\n'
|
|
'Origin: http://127.0.0.1:5678\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n'
|
|
'Sec-WebSocket-Version: 13\r\nSec-WebSocket-Key: '+key+'\r\nCookie: '+cookie+'\r\n\r\n').encode())
|
|
data=b''
|
|
while b'\r\n\r\n' not in data:data+=channel.recv(4096)
|
|
header,buffer=data.split(b'\r\n\r\n',1)
|
|
assert b' 101 ' in header
|
|
accept=base64.b64encode(hashlib.sha1((key+'258EAFA5-E914-47DA-95CA-C5AB0DC85B11').encode()).digest())
|
|
assert accept in header
|
|
def send(value,opcode=1):
|
|
raw=json.dumps(value).encode() if opcode==1 else value;mask=secrets.token_bytes(4)
|
|
prefix=bytes([0x80|opcode,0x80|len(raw)]) if len(raw)<126 else bytes([0x80|opcode,0x80|126])+struct.pack('!H',len(raw))
|
|
channel.sendall(prefix+mask+bytes(c^mask[i%4] for i,c in enumerate(raw)))
|
|
command=('printf "%s\\n" "$PWD" > /home/archipelago/Work/'+nonce+'; '
|
|
'printf ok > /home/archipelago/.local/state/archipelago/sessions/'+nonce+'; '
|
|
'if printf bad > /opt/archipelago/c01-writable-fixture; then printf BAD_WRITE; '
|
|
'else printf DENIED_WRITE; fi; printf "\\n'+marker+'\\n"')
|
|
if os.environ.get('C01_HISTORY_MIGRATION')=='1':
|
|
command=(command.rsplit('printf "\\n'+marker+'\\n"',1)[0]+
|
|
'/usr/bin/podman inspect --format \'{"id":"{{.Id}}","state":{{json .State}}}\' fixture-stopped '
|
|
'> /home/archipelago/Work/'+nonce+'-container; '
|
|
'/usr/bin/podman container exists fixture-removed; '
|
|
'printf "%s\\n" "$?" > /home/archipelago/Work/'+nonce+'-absent; '
|
|
'printf "\\n'+marker+'\\n"')
|
|
yaya_declarations=EVIDENCE/'yaya-container-declarations.json'
|
|
if yaya_declarations.exists():
|
|
declarations=json.loads(yaya_declarations.read_text())['records']
|
|
assert len(declarations)==38 and all(re.fullmatch(r'yaya-declaration-[0-9]+',row['name']) for row in declarations)
|
|
command=(command.rsplit('printf "\\n'+marker+'\\n"',1)[0]+
|
|
'/usr/bin/podman inspect --format \'{"id":"{{.Id}}","state":{{json .State}}}\' '+
|
|
' '.join(row['name'] for row in declarations)+' > /home/archipelago/Work/'+nonce+'-yaya; '+
|
|
'printf "\\n'+marker+'\\n"')
|
|
try:
|
|
send({'type':'input','data':command});send({'type':'input','data':'\r'})
|
|
def take(count):
|
|
nonlocal buffer
|
|
while len(buffer)<count:
|
|
part=channel.recv(65536);assert part,'native terminal EOF';buffer+=part
|
|
assert len(buffer)<=1024*1024
|
|
result,buffer=buffer[:count],buffer[count:];return result
|
|
output='';deadline=time.monotonic()+20
|
|
while time.monotonic()<deadline:
|
|
first,second=take(2);length=second&127
|
|
if length==126:length=struct.unpack('!H',take(2))[0]
|
|
elif length==127:length=struct.unpack('!Q',take(8))[0]
|
|
assert length<=1024*1024 and not second&128
|
|
payload=take(length)
|
|
if first&15==9:send(payload,10);continue
|
|
assert first&15==1,'unexpected native terminal frame'
|
|
message=json.loads(payload);output=message.get('data','')
|
|
# Do not mistake the shell's echoed command for command completion.
|
|
work=Path('/var/lib/c01-fixture/home/Work')/nonce
|
|
session=Path('/var/lib/c01-fixture/home/.local/state/archipelago/sessions')/nonce
|
|
if marker in output and work.exists() and session.exists():
|
|
assert work.read_text()=='/home/archipelago/Work\n' and session.read_text()=='ok'
|
|
assert Path('/opt/archipelago/c01-writable-fixture').read_text()=='retained old-principal-owned fixture\n'
|
|
if os.environ.get('C01_HISTORY_MIGRATION')=='1':
|
|
original=json.loads((EVIDENCE/'migration-seed.json').read_text())
|
|
state=json.loads(work.with_name(nonce+'-container').read_text())
|
|
assert state['id']==original['stopped_container_id']
|
|
assert state['state']['Status'] in ('created','configured') and not state['state']['Running']
|
|
assert state['state']['Pid']==0 and state['state']['StartedAt'].startswith('0001-01-01')
|
|
assert work.with_name(nonce+'-absent').read_text()=='1\n'
|
|
if yaya_declarations.exists():
|
|
actual=[json.loads(line) for line in work.with_name(nonce+'-yaya').read_text().splitlines()]
|
|
assert len(actual)==38
|
|
for observed,expected in zip(actual,declarations):
|
|
assert observed['id']==expected['id'] and observed['state']['Status']=='exited'
|
|
assert observed['state']['Pid']==0 and not observed['state']['Running']
|
|
assert observed['state']['StartedAt']==expected['last_started'],'stopped Yaya projection restarted'
|
|
return {'session':record['id'],'workspace':record['workspace'],'pty_output_sha256':hashlib.sha256(output.encode()).hexdigest(),
|
|
'work_probe':nonce,'denied_package_write':True}
|
|
raise AssertionError('native PTY command did not complete')
|
|
finally:channel.close()
|
|
|
|
|
|
def signature(path):
|
|
m=path.stat();return [m.st_dev,m.st_ino,m.st_uid,m.st_gid,m.st_mode,sha(path)]
|
|
|
|
|
|
def history_fields():
|
|
value=ledger();active=binding();index={k:active[k] for k in ('target_binding_digest','composition_digest','transaction_id')}
|
|
return dict(index,predecessor_epoch_id=value['epoch'],head_sequence=value['sequence'],
|
|
preserved_head_digest=wire.digest(wire.canonical(value['receipt'])) if value['receipt'] else sha(owner.ROOT/'ledger.json'),
|
|
preserved_journal_digest=wire.digest(wire.canonical({'receipts':value['journal'],'intents':value['intents']})),
|
|
preserved_consumed_history_digest=wire.digest(wire.canonical(owner.consumption_history(value))),
|
|
preserved_target_index_digest=wire.digest(wire.canonical(index)))
|
|
|
|
|
|
def external_record(fields):
|
|
directory=PACKAGE/'external-history';directory.mkdir(mode=0o700,exist_ok=True)
|
|
name=secrets.token_hex(16);history=directory/(name+'-history.json')
|
|
store.atomic(history,wire.canonical(dict(fields,sequence=1)))
|
|
record=directory/(name+'-record.json')
|
|
store.atomic(record,wire.canonical(dict(fields,schema_version=1,external_history_sequence=1,
|
|
external_history_digest=sha(history),observed_boot_id=owner.boot_id())))
|
|
return record,history
|
|
|
|
|
|
def main():
|
|
assert os.geteuid()==0 and Path('/run/.containerenv').exists()
|
|
assert socket.gethostname()=='c01-prerequisite-qualification'
|
|
http=NativeHTTP()
|
|
wait(lambda: ledger()['prerequisite']['state']=='STARTED','actual prerequisite observation')
|
|
before_auth=(owner.ROOT/'ledger.json').read_bytes()
|
|
try:http.rpc('update.activate-admitted')
|
|
except urllib.error.HTTPError as error:assert error.code in (401,403)
|
|
else:raise AssertionError('unauthenticated admitted activation accepted')
|
|
assert (owner.ROOT/'ledger.json').read_bytes()==before_auth
|
|
http.login()
|
|
retained={'work':Path('/var/lib/c01-fixture/home/Work/operator.txt'),
|
|
'history':Path('/var/lib/c01-fixture/home/.bash_history'),
|
|
'custom_fragments':Path('/etc/systemd/system/archipelago.service.d/45-retained-operator.conf')}
|
|
if os.environ.get('C01_HISTORY_MIGRATION')=='1':
|
|
launch=json.loads(policy.LAUNCH.read_text());manager=Path(launch['manager_backing']);runtime=Path(launch['runtime_data_dir'])
|
|
retained.update(legacy_history=manager/'federation/pending_requests.json',
|
|
user_stopped=manager/'user-stopped.json',user_uninstalled=manager/'user-uninstalled.json',
|
|
wallet_metadata=manager/'wallet/migration-preserved.bin',runtime_wallet_metadata=runtime/'lnd/migration-preserved.bin',
|
|
removed_data=runtime/'fixture-removed/migration-preserved.bin',subuid_data=runtime/'subuid-preserved.bin')
|
|
yaya_proof=EVIDENCE/'yaya-container-declarations.json'
|
|
if yaya_proof.exists():
|
|
for record in json.loads(yaya_proof.read_text())['records']:
|
|
for index,mount in enumerate(record['retained_mounts']):
|
|
retained['yaya-'+record['name']+'-'+str(index)]=Path(mount['synthetic_source'])/'retained-metadata'
|
|
if os.environ.get('C01_RUNNING_SERVICES')=='1':
|
|
retained.update(fips_public=Path('/etc/fips/fips.pub'),fips_config=Path('/etc/fips/fips.yaml'),
|
|
wg_public=Path('/etc/archipelago-fips-wg/wg.public'),
|
|
tor_public=Path('/var/lib/tor/hidden_service_archipelago/hostname'),
|
|
tor_config=Path('/etc/tor/torrc'),nm_config=Path('/etc/NetworkManager/NetworkManager.conf'))
|
|
before={k:signature(p) for k,p in retained.items()}
|
|
initial=owner.service();assert initial['ActiveState']=='active'
|
|
dump(EVIDENCE/'initial-terminal.json',terminal(http))
|
|
baseline_journal=run('journalctl','-u','archipelago.service','--no-pager').stdout.decode()
|
|
app_pid=re.findall(r'C01_BASELINE_READY application_pid=(\d+)',baseline_journal)[-1]
|
|
baseline_renewals=re.findall(r'C01_CLOCK_REFRESH_COMMITTED application_pid=(\d+) epoch=(\S+)',baseline_journal)
|
|
prior_renewals=baseline_renewals;successful_epochs=set()
|
|
successes=0;refusals=[];deadline=time.monotonic()+400
|
|
while time.monotonic()<deadline:
|
|
assert owner.service()['InvocationID']==initial['InvocationID'],'normal renewal replaced manager'
|
|
read_ok=False
|
|
try:
|
|
http.rpc('federation.list-pending-requests');successes+=1;read_ok=True
|
|
except Exception as error:refusals.append((type(error).__name__+': '+str(error))[:512])
|
|
log=run('journalctl','-u','archipelago.service','--no-pager').stdout.decode()
|
|
renewals=re.findall(r'C01_CLOCK_REFRESH_COMMITTED application_pid=(\d+) epoch=(\S+)',log)
|
|
if read_ok and prior_renewals and renewals and prior_renewals[-1]==renewals[-1]:
|
|
successful_epochs.add(renewals[-1][1])
|
|
normal_renewals=renewals[len(baseline_renewals):]
|
|
if len(normal_renewals)>=2 and all(epoch in successful_epochs for _,epoch in normal_renewals[-2:]):
|
|
assert all(pid==app_pid for pid,_ in renewals) and len({e for _,e in renewals})>=2
|
|
break
|
|
prior_renewals=renewals
|
|
time.sleep(2.3)
|
|
else:
|
|
dump(EVIDENCE/'normal-renewals-refusal.json',{'successful_native_reads':successes,
|
|
'refusals':refusals,'renewals':renewals,'successful_read_epochs':sorted(successful_epochs)})
|
|
raise AssertionError('two post-readiness clock-only renewals with successful reads not observed')
|
|
dump(EVIDENCE/'normal-renewals.json',{'application_pid':app_pid,'manager_invocation':initial['InvocationID'],
|
|
'renewals':normal_renewals,'baseline_renewals':baseline_renewals,
|
|
'successful_read_epochs':sorted(successful_epochs),
|
|
'successful_native_reads':successes,'transient_read_refusals':refusals})
|
|
|
|
def phase(name):
|
|
receipt=ledger().get('receipt');return receipt if receipt and receipt['phase']==name else None
|
|
def exchange(request):
|
|
with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as channel:
|
|
channel.settimeout(5);channel.connect(str(owner.SOCKET));channel.sendall(wire.canonical(request));channel.shutdown(socket.SHUT_WR)
|
|
raw=b''
|
|
while True:
|
|
part=channel.recv(8192)
|
|
if not part:break
|
|
raw+=part;assert len(raw)<=wire.MAX
|
|
return wire.parse(raw)
|
|
def publish(value):
|
|
raw=wire.canonical(value);digest=wire.digest(raw)
|
|
store.atomic(owner.ROOT/'observations'/(digest+'.json'),raw);return digest
|
|
def observe():
|
|
receipt=wait(lambda: phase('AWAITING_OBSERVATION'),'native generation ready',240)
|
|
assert owner.service()['ActiveState']=='active' and owner.service()['InvocationID']==receipt['observed_invocation_id']
|
|
http.login();proof=terminal(http)
|
|
after={k:signature(p) for k,p in retained.items()};assert after==before
|
|
base={'transaction_id':receipt['transaction_id'],'observed_invocation_id':receipt['observed_invocation_id']}
|
|
if os.environ.get('C01_HISTORY_MIGRATION')=='1':
|
|
extended=dict(base,preserved={k:{'before':before[k],'after':after[k]} for k in before})
|
|
proof['migration_preservation_digest']=publish(extended)
|
|
dump(EVIDENCE/('history-preservation-'+receipt['observed_invocation_id']+'.json'),extended)
|
|
namespace=publish(dict(base,qualification='actual native HTTP default terminal and WS PTY commands',
|
|
default_terminal_created=True,default_terminal_workspace=proof['workspace'],
|
|
namespace_writable=['/home/archipelago/Work','/home/archipelago/.local/state/archipelago/sessions'],native_proof=proof))
|
|
preservation=publish(dict(base,preserved={k:{'before':wire.digest(wire.canonical(before[k])),
|
|
'after':wire.digest(wire.canonical(after[k]))} for k in ('work','history','custom_fragments')}))
|
|
observation={k:receipt[k] for k in ('transaction_id','target_binding_digest','composition_digest',
|
|
'observed_invocation_id','effective_config_observation_digest','config_current_digest')}
|
|
observation.update(namespace_observation_digest=namespace,preservation_observation_digest=preservation)
|
|
value=binding()
|
|
request=dict(schema_version=1,operation='ACK_OBSERVATION',route='OBSERVER',reason='OBSERVATION_READY',
|
|
transaction_id=value['transaction_id'],request_id=secrets.token_hex(16),target_binding_digest=value['target_binding_digest'],
|
|
composition_digest=value['composition_digest'],grant_digest=ledger().get('notice_grant') or value['grants']['OBSERVER'],expected_sequence=ledger()['sequence'],
|
|
requester_invocation_id=None,observer_receipt_digest=publish(observation),selected_release_digest=None,staging_generation=None)
|
|
assert exchange(request)['status']=='ACCEPTED'
|
|
return wait(lambda: phase('COMPLETE'),'independent native observation accepted')
|
|
def action(method):
|
|
try:http.rpc(method)
|
|
except (OSError,TimeoutError):pass # The actual owner may retire this HTTP invocation before reply.
|
|
|
|
unchanged=(owner.ROOT/'ledger.json').read_bytes()
|
|
try:http.rpc('update.activate-admitted',{'path':'/unappointed'})
|
|
except RuntimeError:pass
|
|
else:raise AssertionError('caller-selected activation argument accepted')
|
|
assert (owner.ROOT/'ledger.json').read_bytes()==unchanged
|
|
first_binding=copy.deepcopy(binding())
|
|
action('update.activate-admitted');b_receipt=observe()
|
|
assert sha(owner.DESTINATIONS['backend'])==first_binding['artifacts']['backend']
|
|
wait(lambda: run('systemctl','show',ledger()['execution']['unit'],'-p','ActiveState','--value').stdout.strip()==b'inactive','B worker drain')
|
|
# Retry through the new native B process, not a root-injected request. Its
|
|
# retained request must resolve by lookup without another dispatch/grant.
|
|
b_invocation=owner.service()['InvocationID'];b_history=(owner.ROOT/'ledger.json').read_bytes()
|
|
http.rpc('update.activate-admitted')
|
|
assert owner.service()['InvocationID']==b_invocation and (owner.ROOT/'ledger.json').read_bytes()==b_history
|
|
|
|
# Root coordinator supplies a new, history-bound appointment. It cannot
|
|
# reset the old ledger or manufacture an application/observer result.
|
|
incoming=owner.ROOT/'incoming';archive=PACKAGE/'retained-incoming-B';archive.mkdir(mode=0o700)
|
|
for path in incoming.iterdir():path.rename(archive/path.name)
|
|
old_launch=json.loads(policy.LAUNCH.read_text())
|
|
for label in ('new','restore'):
|
|
for name,path in owner.destinations(first_binding).items():
|
|
target=incoming/(label+'-'+name)
|
|
if path.is_dir():shutil.copytree(path,target)
|
|
elif path.exists():shutil.copy2(path,target)
|
|
with (incoming/'new-backend').open('ab') as stream:stream.write(b'\nC01-disposable-generation-C\n')
|
|
new_launch=dict(old_launch,native_sha256=sha(incoming/'new-backend'))
|
|
dump(incoming/'new-launch',new_launch)
|
|
dump(incoming/'exclusion.json',json.loads((policy.STATE/'journal.json').read_text()))
|
|
effective=wire.digest(owner.effective_configuration().encode())
|
|
successor=copy.deepcopy(first_binding)
|
|
successor.pop('requester_uid');successor.pop('requester_gid')
|
|
successor.update(transaction_id=secrets.token_hex(16),admission_epoch_id=secrets.token_hex(16),composition_digest=secrets.token_hex(32),
|
|
grants={k:secrets.token_hex(32) for k in first_binding['grants']},
|
|
artifacts={'backend':sha(incoming/'new-backend'),'frontend':owner.object_digest(incoming/'new-frontend')},
|
|
restore_artifacts=first_binding['artifacts'],effective_config={'APPLY':effective,'RESTORE':effective},
|
|
exclusion_evidence_digest=sha(incoming/'exclusion.json'),
|
|
prerequisite={'grant_digest':secrets.token_hex(32),'backend_sha256':sha(owner.DESTINATIONS['backend']),'effective_config_digest':effective})
|
|
successor['release_identity']=['fixture-C',[['archipelago','fixture-C',successor['artifacts']['backend'],(incoming/'new-backend').stat().st_size]]]
|
|
successor['objects']={name:{'old':owner.object_digest(path),'new':owner.object_digest(incoming/('new-'+name)),
|
|
'restore':owner.object_digest(incoming/('restore-'+name))} for name,path in owner.destinations(successor).items()}
|
|
successor['c01']['generations']={name:{'native_sha256':launch['native_sha256'],'launch_sha256':wire.digest(generation.canonical(launch))}
|
|
for name,launch in [('old',old_launch),('new',new_launch),('restore',old_launch)]}
|
|
for operation,a,b in [('APPLY',old_launch,new_launch),('RESTORE',new_launch,old_launch)]:
|
|
record={'version':1,'policy_sha256':sha(policy.CONFIG),'runtime_sha256':sha(policy.RUNTIME),
|
|
'transaction_id':successor['transaction_id'],'action':operation,'grant_digest':secrets.token_hex(32),
|
|
'from_launch':a,'to_launch':b,'from_launch_sha256':wire.digest(generation.canonical(a)),
|
|
'to_launch_sha256':wire.digest(generation.canonical(b)),
|
|
'from_workspace_digest':successor['objects']['config']['old' if operation=='APPLY' else 'new'],
|
|
'to_workspace_digest':successor['objects']['config']['new' if operation=='APPLY' else 'restore']}
|
|
raw=generation.canonical(record);record['external_history_digest']=wire.digest(raw)
|
|
store.atomic(incoming/('c01-history-'+wire.digest(raw)+'.json'),raw)
|
|
raw=generation.canonical(record);digest=wire.digest(raw)
|
|
store.atomic(incoming/('c01-generation-'+digest+'.json'),raw);successor['c01']['transitions'][operation]=digest
|
|
dump(incoming/'admission.json',successor)
|
|
protected=dict(successor,requester_uid=first_binding['requester_uid'],requester_gid=first_binding['requester_gid'])
|
|
fields=dict(history_fields(),successor_binding_digest=wire.digest(wire.canonical(protected)),terminal_disposition='APPLY')
|
|
record,history=external_record(fields)
|
|
original_journal=list(ledger()['journal']);provision.successor(incoming/'admission.json',record,history)
|
|
assert ledger()['journal']==original_journal
|
|
unchanged=(owner.ROOT/'ledger.json').read_bytes();provision.successor(incoming/'admission.json',record,history)
|
|
assert unchanged==(owner.ROOT/'ledger.json').read_bytes()
|
|
action('update.activate-admitted');c_receipt=observe()
|
|
assert sha(owner.DESTINATIONS['backend'])==successor['artifacts']['backend']
|
|
wait(lambda: run('systemctl','show',ledger()['execution']['unit'],'-p','ActiveState','--value').stdout.strip()==b'inactive','C worker drain')
|
|
action('update.recover-admitted');restored=observe()
|
|
assert restored['config_action']=='RESTORE' and sha(owner.DESTINATIONS['backend'])==first_binding['artifacts']['backend']
|
|
wait(lambda: run('systemctl','show',ledger()['execution']['unit'],'-p','ActiveState','--value').stdout.strip()==b'inactive','restore worker drain')
|
|
previous_invocation=owner.service()['InvocationID'];saved=(owner.ROOT/'ledger.json').read_bytes()
|
|
run('systemctl','restart','archipelago-d09-lifecycle.service')
|
|
wait(lambda: owner.service()['MainPID']=='0' and owner.service()['ActiveState'] in ('inactive','failed'),'owner-loss retirement')
|
|
assert (owner.ROOT/'ledger.json').read_bytes()==saved,'owner restart changed retained history'
|
|
probe=owner.Owner.__new__(owner.Owner)
|
|
drain_waits=[]
|
|
def drained():
|
|
try:return probe.drained_disposition(ledger())
|
|
except ValueError as error:
|
|
reason=str(error)
|
|
if reason not in ('management process/job not drained','prerequisite process still exists',
|
|
'worker descendants not retired','worker process/job not drained','worker process still exists'):
|
|
raise
|
|
if reason not in drain_waits:drain_waits.append(reason)
|
|
return None
|
|
disposition=wait(drained,'owned worker drain proof')
|
|
fields=dict(history_fields(),allowed_action='REACTIVATE_RETAINED',new_epoch_id=secrets.token_hex(16),
|
|
new_grant_digest=secrets.token_hex(32),execution_grant_digest=secrets.token_hex(32),disposition_digest=disposition)
|
|
record,history=external_record(fields);provision.readmit(record,history,activation=True)
|
|
recovered=observe();fresh_invocation=owner.service()['InvocationID']
|
|
assert fresh_invocation!=previous_invocation
|
|
provision.readmit(record,history,activation=True)
|
|
assert owner.service()['InvocationID']==fresh_invocation,'readmission replay dispatched twice'
|
|
dump(EVIDENCE/'normal-generations.json',{'generations':['A','B','C','B'],'B':b_receipt,'C':c_receipt,'restored':restored,
|
|
'owner_recovery':recovered,'recovery_invocation':fresh_invocation,'drain_waits':drain_waits,
|
|
'preservation':before,'journal_entries':len(ledger()['journal']),'consumed_grants':len(ledger()['grants'])})
|
|
if os.environ.get('C01_RUNNING_SERVICES')=='1':
|
|
import c01_admission
|
|
launch=json.loads(policy.LAUNCH.read_text())
|
|
wait(lambda: run('systemctl','show',ledger()['execution']['unit'],'-p','ActiveState','--value').stdout.strip()==b'inactive',
|
|
'final recovery worker drain')
|
|
observations=[]
|
|
# Actual producer loss must retire the application, not retarget a
|
|
# restarted daemon behind its already-issued capability.
|
|
run('systemctl','stop','archipelago-fips.service')
|
|
wait(lambda: owner.service()['MainPID']=='0' and owner.service()['ActiveState'] in ('failed','inactive'),
|
|
'required FIPS producer-loss manager retirement')
|
|
for name,unit in c01_admission.UNITS.items():
|
|
if name!='fips':run('systemctl','stop',unit)
|
|
try:
|
|
contract=c01_admission.Contract(launch)
|
|
except (ValueError,OSError) as error:
|
|
observations.append({'service':name,'unit':unit,'admission_refused':True,'reason':str(error)[:512]})
|
|
else:
|
|
contract.close();raise AssertionError('running service profile accepted stopped '+name)
|
|
run('systemctl','start',unit)
|
|
assert owner.service()['MainPID']=='0','service restoration silently reactivated manager'
|
|
assert {k:signature(p) for k,p in retained.items()}==before
|
|
dump(EVIDENCE/'running-service-loss.json',{'producer_loss_retired_manager':True,
|
|
'automatic_manager_reactivation':False,'refusals':observations,'preservation_unchanged':True})
|
|
return 0
|
|
|
|
|
|
if __name__=='__main__':raise SystemExit(main())
|