Add single-use registration cancellation
This commit is contained in:
parent
752d91cf5c
commit
98250e5838
3 changed files with 71 additions and 6 deletions
|
|
@ -36,3 +36,21 @@ def test_registration_verification_is_digest_only_and_single_use(tmp_path):
|
|||
env["wsgi.input"]=io.BytesIO(consume)
|
||||
replay=json.loads(b"".join(app(env,lambda s,h:result.update(status=s))))
|
||||
assert replay["error"]=="verification_invalid"
|
||||
|
||||
def test_registration_cancellation_is_single_use_and_prevents_verification(tmp_path):
|
||||
provider=Provider(); store=SQLiteDeliveryStore(str(tmp_path/"mail.db")); app=TransactionalApplication(store,provider,"opaque","https://users.example")
|
||||
request={"registration_id":"reg-cancel","normalized_email":"person@example.test","preferred_username":"person","client_id":"coulomb-social","tenant":"tenant:coulomb","correlation_id":"corr-cancel"}
|
||||
raw=json.dumps(request).encode(); result={}
|
||||
app({"PATH_INFO":"/v1/registration-verifications","REQUEST_METHOD":"POST","CONTENT_LENGTH":str(len(raw)),"wsgi.input":io.BytesIO(raw),"HTTP_AUTHORIZATION":"Bearer opaque"},lambda s,h:result.update(status=s))
|
||||
text=provider.calls[0][2]; handle=text.split("handle=",1)[1].splitlines()[0]
|
||||
assert f"/registration/cancel?handle={handle}" in text
|
||||
payload=json.dumps({"handle":handle}).encode()
|
||||
cancel_env={"PATH_INFO":"/v1/registration-verifications/cancel","REQUEST_METHOD":"POST","CONTENT_LENGTH":str(len(payload)),"wsgi.input":io.BytesIO(payload),"HTTP_AUTHORIZATION":"Bearer opaque"}
|
||||
canceled=json.loads(b"".join(app(cancel_env,lambda s,h:result.update(status=s))))
|
||||
assert canceled["purpose"]=="public-registration-cancel"
|
||||
consume_env={"PATH_INFO":"/v1/registration-verifications/consume","REQUEST_METHOD":"POST","CONTENT_LENGTH":str(len(payload)),"wsgi.input":io.BytesIO(payload),"HTTP_AUTHORIZATION":"Bearer opaque"}
|
||||
rejected=json.loads(b"".join(app(consume_env,lambda s,h:result.update(status=s))))
|
||||
assert rejected["error"]=="verification_invalid"
|
||||
cancel_env["wsgi.input"]=io.BytesIO(payload)
|
||||
replay=json.loads(b"".join(app(cancel_env,lambda s,h:result.update(status=s))))
|
||||
assert replay["error"]=="verification_invalid"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue