Spaces:
Running
Running
File size: 3,790 Bytes
0022f23 a136c48 0022f23 a136c48 0022f23 1ec9e51 a136c48 1ec9e51 0022f23 1ec9e51 0022f23 a136c48 0022f23 a136c48 9b989e1 0022f23 9b989e1 0022f23 a136c48 0022f23 1ec9e51 0022f23 1ec9e51 0022f23 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """Verify source revision, JSON APIs, readiness and public read-only boundary."""
import argparse
import json
import subprocess
import sys
from pathlib import Path
import time
import urllib.error
import urllib.request
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--backend-only', action='store_true')
parser.add_argument('--timeout', type=int, default=120)
args = parser.parse_args()
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'backend'))
from app.core.release import source_content_hash
expected_content_hash = source_content_hash()
revision = subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip()
backend = 'https://italianhype-blum.hf.space'
site = 'https://blum.signallayerlabs.workers.dev'
verification_headers = {
'User-Agent': 'BLUM-Deployment-Verification/2.3.0 (+https://github.com/SignalLayerLabs/Blum)',
}
def request(url, **kwargs):
# Identify the maintained verifier explicitly; workers.dev rejects Python's default agent.
return urllib.request.Request(url, headers=verification_headers, **kwargs)
def get_json(origin, path):
response = urllib.request.urlopen(request(origin+path), timeout=15)
if 'application/json' not in response.headers.get('Content-Type', ''):
raise ValueError('Expected a JSON API response: '+path)
return json.load(response)
deadline = time.monotonic() + args.timeout
last_error = ''
while time.monotonic() < deadline:
try:
startup = get_json(backend, '/startup/status')
deployed = get_json(backend, '/api/runtime/release')
if not startup.get('api_ready') or not startup.get('ui_ready') or startup.get('last_error'):
raise ValueError('Backend is restoring or unavailable.')
if deployed.get('source_revision') != revision or deployed.get('version') != '2.3.0':
raise ValueError('Backend revision has not finished deploying.')
if deployed.get('source_content_hash') != expected_content_hash:
raise ValueError('Backend protected source content does not match the verified checkout.')
if not args.backend_only:
public = get_json(site, '/api/release')
if public.get('source_revision') != revision or public.get('source_content_hash') != expected_content_hash or not public.get('backend_ready'):
raise ValueError('Cloudflare revision or backend readiness has not settled.')
break
except (OSError, ValueError) as exc:
last_error = str(exc)
time.sleep(10)
else:
raise SystemExit(('Backend' if args.backend_only else 'Release')+' verification failed: '+last_error)
print('Backend ready at', revision, flush=True)
if not args.backend_only:
for path in ['/api/brain/snapshot', '/api/training/snapshot', '/api/paper-trading/snapshot?limit=1',
'/api/alpha/snapshot', '/api/alpha/desk', '/api/ascent/status']:
payload = get_json(site, path)
if not isinstance(payload, dict): raise SystemExit('Invalid snapshot: '+path)
print('Verified JSON snapshot', path, flush=True)
try:
urllib.request.urlopen(request(site+'/api/ascent/run', data=b'{}', method='POST'), timeout=15)
raise SystemExit('Public research mutation was unexpectedly allowed.')
except urllib.error.HTTPError as exc:
if exc.code != 405: raise
for path in ['/brain/', '/training-ground/', '/paper-trading/', '/alpha/', '/evolution/']:
response = urllib.request.urlopen(request(site+path), timeout=15)
if response.status != 200 or 'text/html' not in response.headers.get('Content-Type', ''):
raise SystemExit('Product page unavailable: '+path)
print('Cloudflare pages, live API snapshots and write protection verified.', flush=True)
|