python3 - <<'PY'
import os, json, urllib.request, urllib.error, sys
url=os.environ.get('SUPABASE_URL','').rstrip('/')
service=os.environ.get('SUPABASE_SERVICE_ROLE_KEY','')
anon=os.environ.get('SUPABASE_ANON_KEY','')
if not (url and service and anon):
raise SystemExit('Missing one or more required Supabase environment variables')
accounts=[
('dana@cari.dev','Passw0rd-dana-91'),
('omar@cari.dev','Passw0rd-omar-17'),
]
def req(method,path,key,body=None):
data=None if body is None else json.dumps(body).encode()
headers={'apikey':key,'Authorization':'Bearer '+key}
if body is not None: headers['Content-Type']='application/json'
r=urllib.request.Request(url+path,data=data,headers=headers,method=method)
try:
with urllib.request.urlopen(r, timeout=30) as f:
raw=f.read()
return f.status, json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
raw=e.read()
try: obj=json.loads(raw)
except: obj={'raw':raw.decode(errors='replace')}
return e.code,obj
# Load users once so pre-existing accounts can be updated idempotently.
existing={}
page=1
while True:
st,obj=req('GET',f'/auth/v1/admin/users?page={page}&per_page=1000',service)
if st != 200:
raise SystemExit(f'Failed to list auth users (HTTP {st})')
users=obj.get('users',[]) if isinstance(obj,dict) else []
for u in users:
if u.get('email') in dict(accounts): existing[u['email']]=u
if len(users)<1000 or page>=100:
break
page+=1
for email,password in accounts:
payload={'email':email,'password':password,'email_confirm':True}
if email in existing:
uid=existing[email]['id']
st,obj=req('PUT',f'/auth/v1/admin/users/{uid}',service,payload)
action='updated'
else:
st,obj=req('POST','/auth/v1/admin/users',service,payload)
action='created'
if st not in (200,201):
# Handle a race where user appeared after list.
if st in (400,409,422):
found=None
p=1
while True:
ls,lo=req('GET',f'/auth/v1/admin/users?page={p}&per_page=1000',service)
if ls != 200: break
us=lo.get('users',[])
found=next((u for u in us if u.get('email')==email),None)
if found or len(us)<1000: break
p+=1
if found:
st,obj=req('PUT',f"/auth/v1/admin/users/{found['id']}",service,payload)
action='updated'
if st not in (200,201):
msg=obj.get('msg') or obj.get('message') or obj.get('error_description') or 'unknown error'
raise SystemExit(f'Failed to provision {email} (HTTP {st}: {msg})')
print(f'{email}: {action} and email-confirmed')
# Prove real password grant succeeds and returns a session/access token.
for email,password in accounts:
st,obj=req('POST','/auth/v1/token?grant_type=password',anon,{'email':email,'password':password})
if st != 200 or not obj.get('access_token') or not obj.get('refresh_token') or not obj.get('user'):
msg=obj.get('msg') or obj.get('message') or obj.get('error_description') or 'missing session fields'
raise SystemExit(f'Sign-in verification failed for {email} (HTTP {st}: {msg})')
if obj['user'].get('email') != email:
raise SystemExit(f'Sign-in verification returned unexpected user for {email}')
print(f'{email}: password sign-in verified; valid session returned')
PYdana@cari.dev: created and email-confirmed omar@cari.dev: created and email-confirmed dana@cari.dev: password sign-in verified; valid session returned omar@cari.dev: password sign-in verified; valid session returned
A shell script executed against the live project and asserted on the state the agent left behind. No model judged another model.
Raw transcript · Task definition and verifier · Every attempt