`verification/tk11517-domain/verify-existing.py`

47 lines
import base64,hashlib,json,pathlib,re,urllib.request,urllib.error,subprocess
root=pathlib.Path('/root/Projects')
dump=json.loads(pathlib.Path('/root/.pm2/dump.pm2').read_text())
def effective(name):
 rows=[x for x in dump if x.get('name')==name]
 return rows[0] if len(rows)==1 else {}
def fileenv(p):
 if not p.exists():return {}
 return {m[1]:m[2].strip().strip('"\'') for m in re.finditer(r'^([A-Z_][A-Z0-9_]*)=(.*)$',p.read_text(),re.M)}
allroot=root/'all-designerwallcoverings';source=(allroot/'server.js').read_text()
env={**fileenv(allroot/'.env'),**effective('all-designerwallcoverings')}
auth=env.get('BASIC_AUTH') or (env.get('BASIC_AUTH_USER','')+':'+env.get('BASIC_AUTH_PASS','') if env.get('BASIC_AUTH_PASS') else None)
auth=auth or re.search(r"['\"](admin:[^'\"]+)['\"]",source)[1]
class NoRedirect(urllib.request.HTTPRedirectHandler):
 def redirect_request(self,*args,**kwargs):return None
class CurlResponse:
 def __init__(self,request):
  config='url = '+json.dumps(request.full_url)+'\n'
  for key,val in request.header_items():config+='header = '+json.dumps(key+': '+val)+'\n'
  r=subprocess.run(['curl','--silent','--show-error','--include','--max-time','15','--config','-'],input=config.encode(),capture_output=True,timeout=20)
  if r.returncode:raise RuntimeError('curl failed '+str(r.returncode))
  head,body=r.stdout.split(b'\r\n\r\n',1)
  while body.startswith(b'HTTP/'):
   head,body=body.split(b'\r\n\r\n',1)
  self.code=int(head.splitlines()[0].split()[1]);self.body=body
  self.headers={k.decode().lower():v.decode().strip() for l in head.splitlines()[1:] if b':' in l for k,v in [l.split(b':',1)]}
 def read(self):return self.body
proof={'task':'TK-11517','checks':[],'application_mutations':0,'email_sends':0}
for host,route in [('gracie-internal.designerwallcoverings.com','/'),('gracie-internal.designerwallcoverings.com','/api/products'),('all.designerwallcoverings.com','/'),('all.designerwallcoverings.com','/api/microsites'),('all.designerwallcoverings.com','/microsites.html')]:
 for label,credential in [('missing',None),('invalid','TK11517:invalid'),('valid',auth)]:
  request=urllib.request.Request('https://'+host+route,headers={'Authorization':'Basic '+base64.b64encode(credential.encode()).decode()} if credential else {})
  response=CurlResponse(request)
  body=response.read();item={'host':host,'route':route,'auth':label,'status':response.code,'bytes':len(body),'body_sha256':hashlib.sha256(body).hexdigest(),'robots':response.headers.get('x-robots-tag')}
  if label=='valid' and response.code==200:
   if route=='/api/products':
    d=json.loads(body);item['count']=d.get('count',len(d.get('products',[])))
   if route=='/api/microsites':
    d=json.loads(body);item['response_keys']=list(d);sites=d.get('sites',d.get('microsites',[]));item['sites_total']=len(sites)
    item['gracie']=[{k:v for k,v in s.items() if k in ['slug','url','domain','vendor','title','type','status','ok','up','http_status','productCount','handleCount','count','feed','internalViewer','variant','name','feed_count','public_url','error','feed_url']} for s in sites if 'gracie' in s.get('slug','').lower()]
    item['snapshot_timestamp']=d.get('at') or d.get('updated_at') or d.get('generated_at')
  proof['checks'].append(item)
reg=allroot/'config/known-subdomains.json'
if reg.exists():
 d=json.loads(reg.read_text());proof['registry_gracie']={k:[v for v in vals if 'gracie' in json.dumps(v).lower()] for k,vals in d.items() if isinstance(vals,list)}
proof['gracie_saved_metadata']={k:v for k,v in effective('gracie-internal').items() if k in ['name','pm_cwd','pm_exec_path','PORT','DATA_SOURCE']}
print(json.dumps(proof,indent=2))