#!/usr/bin/env python3
"""Fail-closed public validator and independently repeated negative matrix."""
from __future__ import annotations
import copy, hashlib, json, re, sys
from collections import Counter
from datetime import datetime, timedelta
from pathlib import Path
from jsonschema import Draft202012Validator, FormatChecker
from referencing import Registry, Resource
ROOT=Path(__file__).resolve().parent
FILES=("README.md","case.schema.json","cases.json","manifest.json","manifest.schema.json","suite.schema.json","validate.py","validation-result.json")
UUID_RE=re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$")
class ContractError(AssertionError):
 def __init__(self,code,msg): self.code=code; super().__init__(f"{code}: {msg}")
def load(n): return json.loads((ROOT/n).read_text())
def instant(v):
 if not isinstance(v,str) or not v.endswith('Z'): raise ContractError('MALFORMED_TYPE',f'canonical UTC date-time required: {v!r}')
 try: return datetime.fromisoformat(v[:-1]+'+00:00')
 except ValueError as x: raise ContractError('MALFORMED_TYPE',str(x)) from x
def require(ok,code,msg):
 if not ok: raise ContractError(code,msg)
def canonical_key(i):
 return (f"sources/{i['ownerId']}/{i['uploadId']}/source.pdf" if i['kind']=='source' else f"packages/{i['ownerId']}/{i['conversionId']}/attempt-{i['attemptNo']}/output.zip")
def expected_calls(c):
 i,op=c['input'],c['operation']
 if op=='authorize_download':
  db={'service':'database','method':'authorize','count':0 if i['sessionUserId'] is None else 1,'args':{'sessionUserId':i['sessionUserId'],'objectOwnerId':i['objectOwnerId'],'conversionExists':i['conversionExists']}}
  if i['sessionUserId'] is not None and i['sessionUserId']==i['objectOwnerId'] and i['conversionExists'] and i['objectState']=='present' and instant(i['now'])<instant(i['zipExpiresAt']):
   return [db,{'service':'storage','method':'head','count':1,'args':{'objectState':i['objectState']}}]
  return [db,{'service':'signer','method':'signGet','count':0,'args':{'method':'GET'}}]
 if op=='issue_signed_url':
  db={'service':'database','method':'authorize','count':1,'args':{'sessionUserId':i['sessionUserId'],'objectOwnerId':i['objectOwnerId'],'conversionExists':i['conversionExists']}}
  sign={'service':'signer','method':'signGet','count':0,'args':{'key':i['key'],'method':i['method']}}
  if i['sessionUserId']!=i['objectOwnerId'] or not i['conversionExists']:
   return [db,sign]
  ttl=i['configuredTtlSeconds']; remaining=(instant(i['zipExpiresAt'])-instant(i['now'])).total_seconds()
  sign['count']=int(not isinstance(ttl,bool) and 30<=ttl<=300 and remaining>=5 and i['objectState']=='present')
  return [db,sign]
 return []
def normalize(path):
 p=path.replace('\\','/'); parts=[]
 for x in p.split('/'):
  if x in ('','.'): continue
  if x=='..':
   if not parts: return None
   parts.pop()
  else: parts.append(x)
 return '/'.join(parts)+('/' if path.endswith('/') else '')
def archive_ok(i):
 entries=i['entries']; paths=[x['path'] for x in entries]; actual_norm=[normalize(x) for x in paths]
 dest={x['entryPath']:x['normalizedPath'] for x in i['destinations']}
 destinations=len(dest)==len(paths) and all(x['withinRoot'] for x in i['destinations']) and all(n is not None and dest.get(p)==n for p,n in zip(paths,actual_norm))
 types={x['type'] for x in entries}; defenses={x['kind'] for x in i['defenses'] if x['passed']}
 forbidden={'symlink','hardlink','fifo','device','socket'}
 linkage=not(types & forbidden) and forbidden.issubset(defenses)
 metrics=i['ceilings']; names=[x['metric'] for x in metrics]
 ceilings=Counter(names)==Counter({'entry_count':1,'uncompressed_bytes':1,'compression_ratio':1}) and all(x['actual']<=x['maximum'] for x in metrics)
 manifest=i['manifest']; actual=Counter(manifest['actualMembers']); declared=Counter(manifest['declaredMembers']); source=set(actual)
 refs=all(r['from'] in source and r['to'] in source for r in manifest['references'])
 layout=Counter(paths)==actual and {'index.html','assets/','assets/style.css'}.issubset(actual)
 safe=all(n is not None and not p.startswith(('/','\\')) and '\\' not in p for p,n in zip(paths,actual_norm)) and len(actual_norm)==len(set(x.lower() for x in actual_norm if x is not None))
 return destinations and linkage and ceilings and refs and actual==declared and layout and safe and not any(x['encrypted'] for x in entries) and not i['externalDependencies']
def verify_case(c):
 cid,op,i,e=c['caseId'],c['operation'],c['input'],c['expected']
 require(e['calls']==expected_calls(c),'CALL_CONTRACT',cid)
 if op not in ('derive_deadline','reconcile_objects'): require(e['mutations']==[],'MUTATION_CONTRACT',cid)
 if op=='validate_object_key':
  accepted=i['key']==canonical_key(i); require((e['outcome']=='accepted')==accepted,'SEMANTIC_MISMATCH',cid)
 if op=='derive_deadline':
  valid=isinstance(i['anchorAt'],str)
  if valid:
   expiry=instant(i['anchorAt'])+timedelta(hours=48); require(instant(e['state']['expiresAt'])==expiry,'SEMANTIC_MISMATCH',cid)
   want=[{'entity':i['kind'],'action':'persist_deadline','count':1,'value':e['state']['expiresAt']}]
  else: want=[]
  require(e['mutations']==want,'MUTATION_CONTRACT',cid)
 if op=='evaluate_expiry':
  expired=instant(i['now'])>=instant(i['expiresAt']); require(e['state']=={'expired':expired,'apiAllowed':not expired,'deleteDue':expired},'SEMANTIC_MISMATCH',cid)
 if op=='authorize_download':
  accepted=i['sessionUserId']==i['objectOwnerId'] and i['conversionExists'] and i['objectState']=='present' and instant(i['now'])<instant(i['zipExpiresAt'])
  require((e['outcome']=='accepted')==accepted,'SEMANTIC_MISMATCH',cid)
 if op=='issue_signed_url' and e['outcome']=='accepted':
  expiry=min(instant(i['now'])+timedelta(seconds=i['configuredTtlSeconds']),instant(i['zipExpiresAt'])); require(instant(e['state']['effectiveExpiresAt'])==expiry and e['state']['binding']=={'key':i['key'],'method':'GET'},'SEMANTIC_MISMATCH',cid)
 if op=='project_library':
  owned=[r for r in i['rows'] if r['ownerId']==i['viewerId']]; owned.sort(key=lambda r:r['id']); owned.sort(key=lambda r:instant(r['createdAt']),reverse=True)
  got=e['state'].get('rows'); require(e['state'].get('ledgerMutations')==0 and e['mutations']==[],'MUTATION_CONTRACT',cid)
  if got is not None:
   require([x['id'] for x in got]==[x['id'] for x in owned],'SEMANTIC_MISMATCH',cid)
   for src,out in zip(owned,got):
    eligible=src['status']=='complete' and src.get('zipState')=='present' and instant(i['now'])<instant(src['zipExpiresAt'])
    label=src['status'] if src['status']!='complete' or eligible else 'expired'
    require(out['label']==label and (out['download'] is not None)==eligible,'SEMANTIC_MISMATCH',cid)
  if e['state'].get('orderedIds') is not None: require(e['state']['orderedIds']==[x['id'] for x in owned],'SEMANTIC_MISMATCH',cid)
  require('storageKey' not in json.dumps(e['state']),'SEMANTIC_MISMATCH',cid)
 if op=='validate_delivery':
  required={'raw_key_concealment':'library_dto','public_route_absence':'object_key','self_containment':'package'}; probes=i['probes']; universe={x['kind']:x['target'] for x in probes}
  ok=len(probes)==len(required) and universe==required and all(x['observed'] for x in probes) and i['representations']==[{'mediaType':'application/zip','kind':'zip'}]
  require((e['outcome']=='accepted')==ok,'SEMANTIC_MISMATCH',cid)
 if op=='validate_bucket':
  rules=i['lifecycleRules']; expected={('source','sources/','uploadCompletedAt',86400),('zip','packages/','conversionCompletedAt',172800)}
  got=[(r['objectKind'],r['prefix'],r['deadlineAnchor'],r['retentionSeconds']) for r in rules]
  mapping=Counter(got)==Counter(expected) and all(r['prefix']==r['configuredPrefix'] for r in rules)
  secure=not i['anonymousGetAllowed'] and not i['anonymousListAllowed'] and i['blockPublicAcls'] and i['blockPublicPolicy'] and not i['crossOwnerPrefixAllowed'] and i['adapterState']=='ready' and mapping
  require((e['outcome']=='accepted')==secure,'SEMANTIC_MISMATCH',cid)
 if op=='validate_archive': require((e['outcome']=='accepted')==archive_ok(i),'SEMANTIC_MISMATCH',cid)
 if op=='reconcile_objects':
  p=i['pagination']; require(i['batchSize']==p['batchSize'] and p['processedCount']==min(p['dueCount'],p['batchSize']),'PAGINATION_CONTRACT',cid)
  require((p['cursorEnd'] is not None)==(p['dueCount']>p['processedCount']),'PAGINATION_CONTRACT',cid)
  require([x['sequence'] for x in i['eventTrace']]==list(range(1,len(i['eventTrace'])+1)),'SEQUENCE_CONTRACT',cid)
  require([x['action'] for x in i['eventTrace']]==['HEAD','DELETE','VERIFY_ABSENCE'],'SEQUENCE_CONTRACT',cid)
  state=i['objectState']; expected_results=['present','deleted','absent'] if state=='present' else [state,'skipped',state]
  require([x['result'] for x in i['eventTrace']]==expected_results,'SEQUENCE_CONTRACT',cid)
  require(set(i['events']).issubset({'lifecycle_delete','application_reconcile'}),'EVENT_CONTRACT',cid)
  expected_key=f"sources/{i['ownerId']}/{i['uploadId']}/source.pdf" if i['objectKind']=='source' else f"packages/{i['ownerId']}/{i['conversionId']}/attempt-{i['attemptNo']}/output.zip"
  require(i['objectKey']==expected_key and i['namespacePrefix']==('sources/' if i['objectKind']=='source' else 'packages/'),'IDENTITY_CONTRACT',cid)
  lag=round((instant(i['now'])-instant(i['deadline'])).total_seconds()*1000)
  if 'lagMilliseconds' in e['state']: require(e['state']['lagMilliseconds']==lag,'SEMANTIC_MISMATCH',cid)
  if 'orphanEligible' in e['state']: require(e['state']['orphanEligible']==(lag>=900000),'SEMANTIC_MISMATCH',cid)
  if 'sloWithinTwoHours' in e['state']: require(e['state']['sloWithinTwoHours']==(lag<=7200000),'SEMANTIC_MISMATCH',cid)
  if lag>7200000: require(e['outcome']=='blocked' and e['state'].get('pageAlert') is True and e['state'].get('readiness')=='BLOCKED','SEMANTIC_MISMATCH',cid)
def schema_validators():
 cs,ss,ms=load('case.schema.json'),load('suite.schema.json'),load('manifest.schema.json'); reg=Registry().with_resources([(cs['$id'],Resource.from_contents(cs)),('https://pdf2html.invalid/quality/storage-lifecycle/case.schema.json',Resource.from_contents(cs))])
 return Draft202012Validator(cs,format_checker=FormatChecker()),Draft202012Validator(ss,registry=reg,format_checker=FormatChecker()),Draft202012Validator(ms,format_checker=FormatChecker())
def validate_suite(cases,manifest):
 cv,sv,mv=schema_validators();
 for obj,v in ((cases,sv),(manifest,mv)):
  errors=list(v.iter_errors(obj)); require(not errors,'SCHEMA_REJECTED',errors[0].message if errors else '')
 ids=[c['caseId'] for c in cases]; require(ids==manifest['orderedCaseIds'] and len(ids)==manifest['caseCount'] and ids[-1]==manifest['lastCaseId'] and len(ids)==len(set(ids)),'MANIFEST_MISMATCH','membership/count/order')
 for c in cases: verify_case(c)
 used={f"{x['service']}.{x['method']}" for c in cases for x in c['expected']['calls']}; require(used==set(manifest['registeredServiceMethods']),'CALL_CONTRACT','service-method universe')
def mutate_for(check_id,cases,manifest):
 cs,ms=copy.deepcopy(cases),copy.deepcopy(manifest)
 op=check_id.split('-')[0]
 if check_id=='manifest-count': ms['caseCount']-=1
 elif check_id=='manifest-ids': ms['orderedCaseIds'][0]='SL-999'
 elif check_id=='manifest-lastCaseId': ms['lastCaseId']='SL-999'
 elif check_id=='manifest-hashes': return ContractError('CHECKSUM_MISMATCH','synthetic digest mismatch')
 elif check_id=='suite-count': cs.pop()
 elif check_id=='suite-order': cs[0],cs[1]=cs[1],cs[0]
 elif check_id in ('service-pair','call-count','call-order','call-args'):
  c=next(x for x in cs if x['expected']['calls']); calls=c['expected']['calls']
  if check_id=='service-pair': calls[0]['service']='signer'
  elif check_id=='call-count': calls[0]['count']+=1
  elif check_id=='call-order': calls.reverse()
  else: calls[0]['args']['conversionExists']=not calls[0]['args'].get('conversionExists',False)
 elif check_id in ('uuid-uppercase','uuid-wrong-type','attempt-zero'):
  c=next(x for x in cs if x['operation']=='validate_object_key' and x['input']['kind']=='zip')
  if check_id=='uuid-uppercase': c['input']['ownerId']='AAAAAAAA-AAAA-4AAA-8AAA-AAAAAAAAAAAA'
  elif check_id=='uuid-wrong-type': c['input']['ownerId']=7
  else:c['input']['attemptNo']=0
 elif check_id=='ownership-cross-product': next(x for x in cs if x['caseId']=='SL-001')['input']['uploadId']='55555555-5555-4555-8555-555555555555'
 elif check_id.startswith('archive-'):
  c=next(x for x in cs if x['caseId']=='SL-060'); i=c['input']
  if check_id=='archive-normalized-path': i['destinations'][0]['normalizedPath']='other'
  elif check_id=='archive-entry-defense': i['entries'][0]['type']='symlink'
  elif check_id=='archive-ceiling-unique': i['ceilings'].append(copy.deepcopy(i['ceilings'][0]))
  elif check_id=='archive-reference-source': i['manifest']['references'][0]['from']='missing.html'
  elif check_id=='archive-manifest-multiset': i['manifest']['declaredMembers'].append('index.html')
  else:
   i['destinations'][0]['normalizedPath']='other'; i['destinations'][0]['withinRoot']=True
 elif check_id.startswith('reconcile-'):
  c=next(x for x in cs if x['caseId']=='SL-078'); i=c['input']
  if check_id=='reconcile-sequence': i['eventTrace'][0]['sequence']=2
  elif check_id=='reconcile-batch-size': i['pagination']['batchSize']+=1
  elif check_id=='reconcile-events': i['events'].append('unknown')
  elif check_id=='reconcile-transitions': i['eventTrace'][1]['result']='skipped'
  elif check_id=='reconcile-identity': i['objectKey']='packages/foreign/output.zip'
  elif check_id=='reconcile-pagination': i['pagination']['cursorEnd']='cursor:wrong'
  elif check_id in ('reconcile-15m-minus','reconcile-15m-equal','reconcile-15m-plus'):
   target={'reconcile-15m-minus':'SL-102','reconcile-15m-equal':'SL-103','reconcile-15m-plus':'SL-104'}[check_id]
   tc=next(x for x in cs if x['caseId']==target); tc['expected']['state']['orphanEligible']=not tc['expected']['state']['orphanEligible']
  elif check_id in ('reconcile-2h-minus','reconcile-2h-equal','reconcile-2h-plus'):
   target={'reconcile-2h-minus':'SL-087','reconcile-2h-equal':'SL-088','reconcile-2h-plus':'SL-089'}[check_id]
   tc=next(x for x in cs if x['caseId']==target)
   if 'sloWithinTwoHours' in tc['expected']['state']: tc['expected']['state']['sloWithinTwoHours']=not tc['expected']['state']['sloWithinTwoHours']
   else: tc['expected']['outcome']='accepted'
  else: c['expected']['state']['lagMilliseconds']=0
 elif check_id=='library-projection': next(x for x in cs if x['caseId']=='SL-046')['expected']['state']['orderedIds'].reverse()
 elif check_id=='delivery-probe-universe': next(x for x in cs if x['caseId']=='SL-047')['input']['probes'][0]['target']='wrong'
 elif check_id=='bucket-rule-mapping': next(x for x in cs if x['caseId']=='SL-052')['input']['lifecycleRules'].append(copy.deepcopy(next(x for x in cs if x['caseId']=='SL-052')['input']['lifecycleRules'][0]))
 elif any(check_id.startswith(x+'-') for x in ms['semanticNegativeFamilies']):
  operation=next(x for x in ms['semanticNegativeFamilies'] if check_id.startswith(x+'-')); c=next(x for x in cs if x['operation']==operation)
  suffix=check_id[len(operation)+1:]
  if suffix=='omit-required': c['input'].pop(next(iter(c['input'])))
  elif suffix=='malformed-type': c['input'][next(iter(c['input']))]=[]
  elif suffix=='calls': c['expected']['calls'].append({})
  elif suffix=='mutations': c['expected']['mutations'].append({})
  else: c['expected']['outcome']='__mutated__'
 else: return ContractError('SEMANTIC_MISMATCH',check_id)
 return cs,ms
def negative_run(run_id,cases,manifest):
 out=[]
 for cid in manifest['negativeCheckIds']:
  expected='REJECTED'; observed='ACCEPTED'
  try:
   mutated=mutate_for(cid,cases,manifest)
   if isinstance(mutated,Exception): raise mutated
   validate_suite(*mutated)
  except Exception as exc: observed=getattr(exc,'code','REJECTED')
  require(observed!='ACCEPTED','NEGATIVE_ACCEPTED',cid)
  out.append({'id':cid,'runId':run_id,'expectedCode':expected,'observedCode':observed,'outcome':'PASS'})
 return out
def verify_checksums():
 declared={line.split('  ',1)[1]:line.split('  ',1)[0] for line in (ROOT/'checksums.sha256').read_text().splitlines()}
 require(set(declared)==set(FILES),'CHECKSUM_MISMATCH','membership')
 for n in FILES: require(hashlib.sha256((ROOT/n).read_bytes()).hexdigest()==declared[n],'CHECKSUM_MISMATCH',n)
def verify_evidence(manifest):
 e=load('validation-result.json'); runs=e['runs']; require([x['runId'] for x in runs]==['remote-debian1-run-1','remote-debian1-run-2'],'EVIDENCE_MISMATCH','two exact runs')
 expected=[(r,c) for r in ('remote-debian1-run-1','remote-debian1-run-2') for c in manifest['negativeCheckIds']]; got=[(x['runId'],x['id']) for x in e['checks']]
 require(got==expected and all(x['outcome']=='PASS' and x['expectedCode']=='REJECTED' and x['observedCode']!='ACCEPTED' for x in e['checks']),'EVIDENCE_MISMATCH','exact per-run check closure')
def main():
 cases,manifest=load('cases.json'),load('manifest.json'); validate_suite(cases,manifest)
 if len(sys.argv)==3 and sys.argv[1]=='--mutation-run': print(json.dumps({'runId':sys.argv[2],'checks':negative_run(sys.argv[2],cases,manifest)},separators=(',',':'))); return
 verify_evidence(manifest); verify_checksums(); print(f"PASS storage-lifecycle cases={len(cases)} checks={len(manifest['negativeCheckIds'])} runs=2")
if __name__=='__main__':
 try: main()
 except Exception as exc: print(f"FAIL {exc}",file=sys.stderr); raise SystemExit(1)