coulomb/email-connect
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add single-use registration verification
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 2s
Details

Browse source
This commit is contained in:
tegwick 2026-08-10 12:29:55 +02:00
parent be5aab1297
commit 752d91cf5c
3 changed files with 86 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

15
tests/test_transactional.py
View file

@ -21,3 +21,18 @@ def test_rejects_auth_recipient_template_and_key(tmp_path):
bad=event(); bad["data"]["primary_email"]="bad\n@example.test"; assert invoke(app,bad)[1]["error"]=="invalid_recipient"
bad=event(); bad["type"]="arbitrary.send"; assert invoke(app,bad)[1]["error"]=="template_not_allowed"
assert invoke(app,event(),key="other")[1]["error"]=="idempotency_key_mismatch"
def test_registration_verification_is_digest_only_and_single_use(tmp_path):
provider=Provider(); store=SQLiteDeliveryStore(str(tmp_path/"mail.db")); app=TransactionalApplication(store,provider,"opaque","https://users.example")
request={"registration_id":"reg-1","normalized_email":"person@example.test","preferred_username":"person","client_id":"coulomb-social","tenant":"tenant:coulomb","correlation_id":"corr-1"}
raw=json.dumps(request).encode(); result={}; response=json.loads(b"".join(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))))
assert result["status"].startswith("202") and response["accepted"]
text=provider.calls[0][2]; handle=text.split("handle=",1)[1].splitlines()[0]
stored=store.db.execute("SELECT handle_hash FROM verifications").fetchone()[0]
assert handle not in stored and stored
consume=json.dumps({"handle":handle}).encode(); env={"PATH_INFO":"/v1/registration-verifications/consume","REQUEST_METHOD":"POST","CONTENT_LENGTH":str(len(consume)),"wsgi.input":io.BytesIO(consume),"HTTP_AUTHORIZATION":"Bearer opaque"}
evidence=json.loads(b"".join(app(env,lambda s,h:result.update(status=s))))
assert evidence["purpose"]=="public-registration" and evidence["email"]=="person@example.test"
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"