121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the exact shared route suite on the pinned disposable Forgejo image."""
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import uuid
|
|
|
|
HERE = Path(__file__).parent
|
|
|
|
|
|
def load(name, filename):
|
|
spec = importlib.util.spec_from_file_location(name, HERE / filename)
|
|
module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
base = load('lifecycle', 'verify_forge_lifecycle.py')
|
|
routes = load('routes', 'accept_forge_routes.py')
|
|
LOADED_ROUTES_SHA256 = hashlib.sha256((HERE / 'accept_forge_routes.py').read_bytes()).hexdigest()
|
|
|
|
|
|
class FixtureReady(Exception):
|
|
pass
|
|
|
|
|
|
class SetupProbe(base.Probe):
|
|
def check(self, name, condition):
|
|
super().check(name, condition)
|
|
if name == 'native_repository_created':
|
|
raise FixtureReady()
|
|
|
|
|
|
class Backend:
|
|
user = 'probe'
|
|
origin = 'http://127.0.0.1:3000'
|
|
scope = 'Exact shared HTTP/API suite on pinned isolated Forgejo, SQLite, no external network; no live credential or deployed acceptance.'
|
|
|
|
def __init__(self, probe):
|
|
self.p = probe
|
|
self.directory = base.ROOT + '/routes-client'
|
|
probe.execute('mkdir', self.directory)
|
|
|
|
def api(self, method, path, data=None, authenticated=True):
|
|
return self.p.api(method, path, data, authenticated)
|
|
|
|
def git(self, *args, allow_failure=False):
|
|
auth = base64.b64encode(('probe:' + self.p.password).encode()).decode()
|
|
return self.p.execute('env', 'GIT_CONFIG_COUNT=1',
|
|
'GIT_CONFIG_KEY_0=http.http://127.0.0.1:3000/.extraHeader',
|
|
'GIT_CONFIG_VALUE_0=Authorization: Basic ' + auth,
|
|
'git', '-C', self.directory, *args,
|
|
allowed=(0, 1, 128) if allow_failure else (0,), timeout=120)
|
|
|
|
def put(self, name, content):
|
|
self.p.put(self.directory + '/' + name, content)
|
|
|
|
def fixture(self):
|
|
return self.p.execute('cat', base.ROOT + '/synthetic-key').stdout
|
|
|
|
def profile(self, name):
|
|
path = self.p.profile(name)
|
|
digest = self.p.execute('sha256sum', path).stdout.decode().split()[0]
|
|
mode = self.p.execute('stat', '-c', '%a %u %g', path).stdout.decode().strip()
|
|
self.p.evidence['inherited_hook_mode_owner'] = mode
|
|
if digest != routes.PROFILE['hook_sha256'] or mode not in ('555 1000 1000', '755 1000 1000'):
|
|
raise ValueError('inherited-profile-mismatch')
|
|
|
|
def absent_object(self, name, oid):
|
|
root = base.REPOS + '/' + name + '.git'
|
|
raw = self.p.execute('git', '--git-dir=' + root, 'cat-file', '--batch-check',
|
|
payload=(oid + '\n').encode()).stdout.decode().strip()
|
|
if raw != oid + ' missing':
|
|
raise ValueError('quarantine-retained')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--artifact-dir', type=Path, required=True)
|
|
parser.add_argument('--receipt', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
if args.receipt.exists():
|
|
raise SystemExit('Retain prior attempt; use a new receipt.')
|
|
probe = SetupProbe(args.receipt.with_suffix('.baseline.json'))
|
|
suite = None
|
|
try:
|
|
try:
|
|
probe.run_probe(args.artifact_dir)
|
|
except FixtureReady:
|
|
pass
|
|
probe.evidence['scope'] = 'Pinned-image fixture setup only; shared suite has its own acceptance receipt.'
|
|
template = base.ROOT + '/template'
|
|
probe.execute('mkdir', '-p', template + '/hooks/pre-receive.d')
|
|
probe.put(template + '/hooks/pre-receive.d/10-custodian-secret-scan',
|
|
(HERE / 'receive-profile-hook').read_bytes(), '555')
|
|
probe.execute('git', 'config', '--file', base.ROOT + '/git-home/.gitconfig', 'init.templateDir', template)
|
|
probe.execute('ssh-keygen', '-q', '-t', 'rsa', '-b', '2048', '-N', '', '-f', base.ROOT + '/synthetic-key')
|
|
probe.evidence['result'] = 'fixture-ready'
|
|
suite = routes.Suite(Backend(probe), args.receipt, 'custodian-routes-' + uuid.uuid4().hex[:24])
|
|
suite.evidence['loaded_executor_sha256'] = LOADED_ROUTES_SHA256
|
|
suite.run()
|
|
suite.check('executor_unchanged_during_rehearsal', LOADED_ROUTES_SHA256 ==
|
|
hashlib.sha256((HERE / 'accept_forge_routes.py').read_bytes()).hexdigest())
|
|
except Exception:
|
|
if suite:
|
|
suite.evidence['failure_phase'] = suite.phase
|
|
raise SystemExit('Shared route acceptance incomplete; raw diagnostics suppressed.')
|
|
finally:
|
|
try:
|
|
if suite:
|
|
suite.close()
|
|
finally:
|
|
probe.close()
|
|
print(json.dumps({'result': suite.evidence['result'], 'checks': len(suite.evidence['checks']),
|
|
'receipt': str(args.receipt)}))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|