`verification/crezana-proposal/refresh.py`

36 lines
import pathlib,json,hashlib,subprocess,datetime,urllib.request,urllib.error,base64,re
base=pathlib.Path('/Users/macstudio3/Projects/tk11438-postgres-migration')
ledger=json.loads((base/'verification/fleet-classification/patch-manifest.json').read_text())
raw=subprocess.check_output(['pm2','jlist'],text=True,timeout=30)
live=json.loads(raw[raw.index('['):])
saved=json.loads(pathlib.Path('/Users/macstudio3/.pm2/dump.pm2').read_text())
def safe(env):
 return {k:env[k] for k in ['PGHOST','PGPORT','PGDATABASE','PGUSER','PORT','DATA_SOURCE'] if k in env}
results=[]
for m in ledger['modules']:
 p=pathlib.Path(m['path']);root=p.parent.parent
 h=hashlib.sha256(p.read_bytes()).hexdigest()
 matches=[a for a in live if a.get('pm2_env',{}).get('pm_cwd')==str(root)]
 names=[a['name'] for a in matches]
 results.append({'project':m['project'],'source':str(p),'sha256':h,'patch_state':'prepared-not-applied' if h==m['before_sha256'] else 'applied' if h==m['after_sha256'] else 'drift-review-required','head':subprocess.check_output(['git','rev-parse','HEAD'],cwd=root,text=True).strip(),'dirty_paths':subprocess.check_output(['git','status','--porcelain'],cwd=root,text=True).splitlines(),'runtime':[{'name':a['name'],'pid':a['pid'],'status':a['pm2_env'].get('status'),'restarts':a['pm2_env'].get('restart_time'),'env':safe(a['pm2_env'])} for a in matches],'saved':[{'name':a['name'],'env':safe(a)} for a in saved if a.get('name') in names]})
crezana=next(r for r in results if r['project']=='crezana-internal')
server=pathlib.Path('/Users/macstudio3/Projects/crezana-internal/server.js').read_text()
user=re.search(r"cred\.name !== '([^']+)'",server).group(1)
password=re.search(r"cred\.pass !== '([^']+)'",server).group(1)
auth='Basic '+base64.b64encode((user+':'+password).encode()).decode()
checks=[]
for path,mode,expected in [('/', 'missing',401),('/','invalid',401),('/','valid',200),('/api/products','missing',200),('/api/products','valid',200),('/api/health','missing',200)]:
 headers={} if mode=='missing' else {'Authorization':auth if mode=='valid' else 'Basic aW52YWxpZDppbnZhbGlk'}
 req=urllib.request.Request('http://127.0.0.1:10072'+path,headers=headers)
 try:
  with urllib.request.urlopen(req,timeout=15) as response: status=response.status;body=response.read()
 except urllib.error.HTTPError as e: status=e.code;body=e.read()
 item={'path':path,'auth':mode,'status':status,'expected':expected,'pass':status==expected,'body_sha256':hashlib.sha256(body).hexdigest()}
 if path=='/api/products' and status==200:
  data=json.loads(body);item['response_shape']=list(data) if isinstance(data,dict) else 'array'
  item['total']=data.get('total',data.get('count')) if isinstance(data,dict) else len(data)
 checks.append(item)
out={'at':datetime.datetime.now(datetime.timezone.utc).isoformat(),'scope':'Read-only source/runtime refresh and Crezana HTTP baseline; no module imports, requests API calls, DB writes, restarts or emails','modules':results,'crezana_http':checks,'crezana_server_sha256':hashlib.sha256(server.encode()).hexdigest(),'all_http_pass':all(x['pass'] for x in checks)}
pathlib.Path('/tmp/tk11438-resume-preflight.json').write_text(json.dumps(out,indent=2)+'\n')
print(json.dumps({'states':{s:sum(r['patch_state']==s for r in results) for s in {r['patch_state'] for r in results}},'crezana':crezana,'http':checks},indent=2))