63 lines
3.6 KiB
Python
63 lines
3.6 KiB
Python
"""Explicit Linux/Docker smoke test: disposable container, no published ports."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
|
|
NAME='athena-deck-installer-smoke'
|
|
IMAGE='athena-deck-installer-test:local'
|
|
ROOT=Path(__file__).resolve().parent.parent
|
|
|
|
def run(*args,data=None,check=True):
|
|
result=subprocess.run(args,input=data,text=True,capture_output=True,timeout=120)
|
|
if check and result.returncode:raise RuntimeError('Smoke command failed (output withheld): '+args[0])
|
|
return result.stdout.strip()
|
|
|
|
def main():
|
|
if run('docker','ps','-aq','--filter','name=^/'+NAME+'$'):
|
|
raise RuntimeError('Smoke container already exists; refusing takeover')
|
|
ids=run('docker','ps','-aq').splitlines()
|
|
before=run('docker','inspect','--format','{{.Id}} {{.State.StartedAt}}',*ids) if ids else ''
|
|
with tempfile.TemporaryDirectory(prefix='deck-install-smoke-') as directory:
|
|
base=Path(directory)
|
|
for name in ('state','bootstrap'):
|
|
(base/name).mkdir(mode=0o700);os.chown(base/name,65534,65534)
|
|
password=secrets.token_urlsafe(32)
|
|
script="""import json,sys,runpy
|
|
from unittest.mock import patch
|
|
p=json.load(sys.stdin)['password'];values=iter([p,p,''])
|
|
with patch('getpass.getpass',side_effect=lambda _:next(values)),patch('sys.stdin.isatty',return_value=True):runpy.run_path('/provision.py',run_name='__main__')
|
|
"""
|
|
mounts=['-v',str(base/'state')+':/var/lib/deck','-v',str(base/'bootstrap')+':/bootstrap','-v',str(ROOT/'deploy/provision.py')+':/provision.py:ro']
|
|
run('docker','run','--rm','-i','--network','none','--user','65534:65534','--cap-drop','ALL','--read-only','-e','PYTHONPATH=/app',*mounts,IMAGE,'python3','-c',script,data=json.dumps({'password':password}))
|
|
token=(base/'bootstrap/api-token.txt').read_text().strip()
|
|
assert (base/'bootstrap/api-token.txt').stat().st_mode&0o777==0o600
|
|
assert token not in (base/'state/auth.json').read_text()
|
|
print('PASS interactive provisioner with synthetic inputs; token protected, hashes persisted',flush=True)
|
|
try:
|
|
run('docker','run','-d','--name',NAME,'--network','none','--read-only','--cap-drop','ALL','--security-opt','no-new-privileges:true','--tmpfs','/tmp:rw,nosuid,nodev,size=8m','-v',str(base/'state')+':/var/lib/deck',IMAGE)
|
|
probe="""import json,sys,urllib.request
|
|
v=json.load(sys.stdin)
|
|
req=urllib.request.Request('http://127.0.0.1:8108/api/v1/status',headers={'Authorization':'Bearer '+v['token']})
|
|
with urllib.request.urlopen(req,timeout=3) as r:assert json.load(r)['location']=='Server'
|
|
"""
|
|
for _ in range(20):
|
|
try:run('docker','exec','-i',NAME,'python3','-c',probe,data=json.dumps({'token':token}));break
|
|
except RuntimeError:time.sleep(.5)
|
|
else:raise RuntimeError('Server not ready')
|
|
run('docker','stop',NAME);run('docker','start',NAME)
|
|
for _ in range(20):
|
|
try:run('docker','exec','-i',NAME,'python3','-c',probe,data=json.dumps({'token':token}));break
|
|
except RuntimeError:time.sleep(.5)
|
|
else:raise RuntimeError('Server not ready after restart')
|
|
print('PASS unprivileged standalone server, API authentication and persistence after restart',flush=True)
|
|
after=run('docker','inspect','--format','{{.Id}} {{.State.StartedAt}}',*ids) if ids else ''
|
|
assert before==after
|
|
print('PASS previously existing container start times unchanged',flush=True)
|
|
finally:run('docker','rm','-f',NAME,check=False)
|
|
|
|
if __name__=='__main__':main()
|