feat: implement Story 5.1 - Magic Link Request
Allow participants to request magic links for dashboard access: - POST /exchange/<slug>/request-access handles form submission - Accept email, look up participant in database - Generate token (secrets.token_urlsafe(32)), store SHA-256 hash - Send magic link email via EmailService.send_magic_link() - Rate limit: 3 requests per hour per email - Always show generic success message (prevent enumeration) - Only send email if participant exists - Case-insensitive email lookup - Comprehensive test suite with 7 tests Includes: - MagicLinkRequestForm with email validation - request_access.html template - GET endpoint to display request form All tests passing (97 total), 91% coverage maintained. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -35,3 +35,16 @@ class ParticipantRegistrationForm(FlaskForm):
|
|||||||
default=True,
|
default=True,
|
||||||
description="Receive email reminders about important dates",
|
description="Receive email reminders about important dates",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class MagicLinkRequestForm(FlaskForm):
|
||||||
|
"""Form for requesting a magic link."""
|
||||||
|
|
||||||
|
email = EmailField(
|
||||||
|
"Email",
|
||||||
|
validators=[
|
||||||
|
DataRequired(message="Email is required"),
|
||||||
|
Email(message="Please enter a valid email address"),
|
||||||
|
Length(max=255, message="Email must be less than 255 characters"),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ from datetime import UTC, datetime, timedelta
|
|||||||
from flask import Blueprint, abort, flash, redirect, render_template, request, url_for
|
from flask import Blueprint, abort, flash, redirect, render_template, request, url_for
|
||||||
|
|
||||||
from src.app import db
|
from src.app import db
|
||||||
from src.forms.participant import ParticipantRegistrationForm
|
from src.forms.participant import MagicLinkRequestForm, ParticipantRegistrationForm
|
||||||
from src.models.exchange import Exchange
|
from src.models.exchange import Exchange
|
||||||
from src.models.magic_token import MagicToken
|
from src.models.magic_token import MagicToken
|
||||||
from src.models.participant import Participant
|
from src.models.participant import Participant
|
||||||
@@ -166,6 +166,102 @@ def register_success(slug: str):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@participant_bp.route("/exchange/<slug>/request-access", methods=["GET", "POST"])
|
||||||
|
def request_access(slug: str):
|
||||||
|
"""Request a magic link for participant access.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
slug: Exchange registration slug.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Rendered request access page template or success message.
|
||||||
|
"""
|
||||||
|
# Find the exchange by slug
|
||||||
|
exchange = db.session.query(Exchange).filter_by(slug=slug).first()
|
||||||
|
if not exchange:
|
||||||
|
abort(404)
|
||||||
|
|
||||||
|
# Create the magic link request form
|
||||||
|
form = MagicLinkRequestForm()
|
||||||
|
|
||||||
|
# Handle POST request
|
||||||
|
if form.validate_on_submit():
|
||||||
|
# Lowercase email for consistency
|
||||||
|
email = form.email.data.lower()
|
||||||
|
|
||||||
|
# Rate limiting: 3 requests per hour per email
|
||||||
|
rate_limit_key = f"magic_link:{slug}:{email}"
|
||||||
|
|
||||||
|
if check_rate_limit(rate_limit_key, max_attempts=3, window_minutes=60):
|
||||||
|
abort(429) # Too Many Requests
|
||||||
|
|
||||||
|
# Look up participant
|
||||||
|
participant = (
|
||||||
|
db.session.query(Participant)
|
||||||
|
.filter_by(exchange_id=exchange.id, email=email)
|
||||||
|
.first()
|
||||||
|
)
|
||||||
|
|
||||||
|
# Always show generic success message (prevent enumeration)
|
||||||
|
# But only send email if participant exists
|
||||||
|
if participant:
|
||||||
|
# Generate magic token
|
||||||
|
token = secrets.token_urlsafe(32)
|
||||||
|
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||||||
|
|
||||||
|
magic_token = MagicToken(
|
||||||
|
token_hash=token_hash,
|
||||||
|
token_type="magic_link",
|
||||||
|
email=email,
|
||||||
|
participant_id=participant.id,
|
||||||
|
exchange_id=exchange.id,
|
||||||
|
expires_at=datetime.now(UTC) + timedelta(hours=1),
|
||||||
|
)
|
||||||
|
magic_token.validate()
|
||||||
|
db.session.add(magic_token)
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
# Send magic link email
|
||||||
|
magic_link_url = url_for(
|
||||||
|
"participant.magic_login",
|
||||||
|
token=token,
|
||||||
|
_external=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
email_service = EmailService()
|
||||||
|
try:
|
||||||
|
email_service.send_magic_link(
|
||||||
|
to=email,
|
||||||
|
magic_link_url=magic_link_url,
|
||||||
|
exchange_name=exchange.name,
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
# Log error but don't fail the request
|
||||||
|
print(f"Failed to send magic link email: {e}")
|
||||||
|
|
||||||
|
# Increment rate limit
|
||||||
|
increment_rate_limit(rate_limit_key, window_minutes=60)
|
||||||
|
|
||||||
|
# Show generic success message
|
||||||
|
flash(
|
||||||
|
"If you're registered for this exchange, we've sent you a magic link. "
|
||||||
|
"Please check your email.",
|
||||||
|
"success",
|
||||||
|
)
|
||||||
|
return render_template(
|
||||||
|
"participant/request_access.html",
|
||||||
|
exchange=exchange,
|
||||||
|
form=form,
|
||||||
|
success=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
return render_template(
|
||||||
|
"participant/request_access.html",
|
||||||
|
exchange=exchange,
|
||||||
|
form=form,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@participant_bp.route("/auth/participant/magic/<token>")
|
@participant_bp.route("/auth/participant/magic/<token>")
|
||||||
def magic_login(token: str): # noqa: ARG001
|
def magic_login(token: str): # noqa: ARG001
|
||||||
"""Magic link login for participants.
|
"""Magic link login for participants.
|
||||||
|
|||||||
38
src/templates/participant/request_access.html
Normal file
38
src/templates/participant/request_access.html
Normal file
@@ -0,0 +1,38 @@
|
|||||||
|
{% extends "layouts/base.html" %}
|
||||||
|
|
||||||
|
{% block title %}Request Access - {{ exchange.name }}{% endblock %}
|
||||||
|
|
||||||
|
{% block content %}
|
||||||
|
<article>
|
||||||
|
<header>
|
||||||
|
<h1>Request Access</h1>
|
||||||
|
<p>{{ exchange.name }}</p>
|
||||||
|
</header>
|
||||||
|
|
||||||
|
{% if success %}
|
||||||
|
<section>
|
||||||
|
<p><strong>Check your email!</strong></p>
|
||||||
|
<p>If you're registered for this exchange, we've sent you a magic link to access your participant dashboard.</p>
|
||||||
|
<p>The link will expire in 1 hour.</p>
|
||||||
|
</section>
|
||||||
|
{% else %}
|
||||||
|
<section>
|
||||||
|
<p>Enter your email address to receive a magic link for access to your participant dashboard.</p>
|
||||||
|
|
||||||
|
<form method="POST">
|
||||||
|
{{ form.hidden_tag() }}
|
||||||
|
|
||||||
|
<label for="email">
|
||||||
|
{{ form.email.label }}
|
||||||
|
{{ form.email(placeholder="your.email@example.com") }}
|
||||||
|
{% if form.email.errors %}
|
||||||
|
<small>{{ form.email.errors[0] }}</small>
|
||||||
|
{% endif %}
|
||||||
|
</label>
|
||||||
|
|
||||||
|
<button type="submit">Send Magic Link</button>
|
||||||
|
</form>
|
||||||
|
</section>
|
||||||
|
{% endif %}
|
||||||
|
</article>
|
||||||
|
{% endblock %}
|
||||||
241
tests/integration/test_story_5_1_magic_link_request.py
Normal file
241
tests/integration/test_story_5_1_magic_link_request.py
Normal file
@@ -0,0 +1,241 @@
|
|||||||
|
"""Tests for Story 5.1: Magic Link Request.
|
||||||
|
|
||||||
|
As a registered participant, I want to request a magic link
|
||||||
|
so that I can access my participant dashboard.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
from src.models.exchange import Exchange
|
||||||
|
from src.models.magic_token import MagicToken
|
||||||
|
from src.models.participant import Participant
|
||||||
|
|
||||||
|
|
||||||
|
class TestStory51MagicLinkRequest:
|
||||||
|
"""Test suite for Story 5.1: Magic Link Request."""
|
||||||
|
|
||||||
|
def test_request_magic_link_for_existing_participant(self, client, db):
|
||||||
|
"""Test successful magic link request for existing participant."""
|
||||||
|
future_close_date = datetime.utcnow() + timedelta(days=7)
|
||||||
|
future_exchange_date = datetime.utcnow() + timedelta(days=14)
|
||||||
|
|
||||||
|
exchange = Exchange(
|
||||||
|
slug="winter2025",
|
||||||
|
name="Office Secret Santa 2025",
|
||||||
|
budget="$50",
|
||||||
|
max_participants=10,
|
||||||
|
registration_close_date=future_close_date,
|
||||||
|
exchange_date=future_exchange_date,
|
||||||
|
timezone="America/New_York",
|
||||||
|
state=Exchange.STATE_REGISTRATION_OPEN,
|
||||||
|
)
|
||||||
|
db.session.add(exchange)
|
||||||
|
db.session.flush()
|
||||||
|
|
||||||
|
participant = Participant(
|
||||||
|
exchange_id=exchange.id,
|
||||||
|
name="John Doe",
|
||||||
|
email="john@example.com",
|
||||||
|
reminder_enabled=True,
|
||||||
|
)
|
||||||
|
db.session.add(participant)
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
# Request magic link
|
||||||
|
response = client.post(
|
||||||
|
"/exchange/winter2025/request-access",
|
||||||
|
data={"email": "john@example.com"},
|
||||||
|
follow_redirects=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Should show success message
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert b"check your email" in response.data or b"sent you" in response.data
|
||||||
|
|
||||||
|
# Should create magic token
|
||||||
|
token = (
|
||||||
|
db.session.query(MagicToken)
|
||||||
|
.filter_by(participant_id=participant.id)
|
||||||
|
.first()
|
||||||
|
)
|
||||||
|
assert token is not None
|
||||||
|
assert token.email == "john@example.com"
|
||||||
|
assert token.token_type == "magic_link"
|
||||||
|
|
||||||
|
def test_request_magic_link_case_insensitive_email(self, client, db):
|
||||||
|
"""Test that email lookup is case insensitive."""
|
||||||
|
future_close_date = datetime.utcnow() + timedelta(days=7)
|
||||||
|
future_exchange_date = datetime.utcnow() + timedelta(days=14)
|
||||||
|
|
||||||
|
exchange = Exchange(
|
||||||
|
slug="test123",
|
||||||
|
name="Test Exchange",
|
||||||
|
budget="$30",
|
||||||
|
max_participants=10,
|
||||||
|
registration_close_date=future_close_date,
|
||||||
|
exchange_date=future_exchange_date,
|
||||||
|
timezone="America/New_York",
|
||||||
|
state=Exchange.STATE_REGISTRATION_OPEN,
|
||||||
|
)
|
||||||
|
db.session.add(exchange)
|
||||||
|
db.session.flush()
|
||||||
|
|
||||||
|
participant = Participant(
|
||||||
|
exchange_id=exchange.id,
|
||||||
|
name="Jane Smith",
|
||||||
|
email="jane.smith@example.com",
|
||||||
|
reminder_enabled=True,
|
||||||
|
)
|
||||||
|
db.session.add(participant)
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
# Request with different case
|
||||||
|
response = client.post(
|
||||||
|
"/exchange/test123/request-access",
|
||||||
|
data={"email": "Jane.Smith@Example.COM"},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Should create token for lowercased email
|
||||||
|
token = (
|
||||||
|
db.session.query(MagicToken)
|
||||||
|
.filter_by(email="jane.smith@example.com")
|
||||||
|
.first()
|
||||||
|
)
|
||||||
|
assert token is not None
|
||||||
|
|
||||||
|
def test_request_magic_link_for_nonexistent_email(self, client, db):
|
||||||
|
"""Test magic link request for email not registered.
|
||||||
|
|
||||||
|
Should prevent enumeration attacks by showing success message.
|
||||||
|
"""
|
||||||
|
future_close_date = datetime.utcnow() + timedelta(days=7)
|
||||||
|
future_exchange_date = datetime.utcnow() + timedelta(days=14)
|
||||||
|
|
||||||
|
exchange = Exchange(
|
||||||
|
slug="test123",
|
||||||
|
name="Test Exchange",
|
||||||
|
budget="$30",
|
||||||
|
max_participants=10,
|
||||||
|
registration_close_date=future_close_date,
|
||||||
|
exchange_date=future_exchange_date,
|
||||||
|
timezone="America/New_York",
|
||||||
|
state=Exchange.STATE_REGISTRATION_OPEN,
|
||||||
|
)
|
||||||
|
db.session.add(exchange)
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
# Request for email that doesn't exist
|
||||||
|
response = client.post(
|
||||||
|
"/exchange/test123/request-access",
|
||||||
|
data={"email": "nonexistent@example.com"},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Should still show generic success message (prevent enumeration)
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert b"check your email" in response.data or b"sent you" in response.data
|
||||||
|
|
||||||
|
# Should NOT create a token
|
||||||
|
token = (
|
||||||
|
db.session.query(MagicToken)
|
||||||
|
.filter_by(email="nonexistent@example.com")
|
||||||
|
.first()
|
||||||
|
)
|
||||||
|
assert token is None
|
||||||
|
|
||||||
|
def test_request_magic_link_rate_limiting(self, client, db):
|
||||||
|
"""Test rate limiting for magic link requests (3 per hour per email)."""
|
||||||
|
future_close_date = datetime.utcnow() + timedelta(days=7)
|
||||||
|
future_exchange_date = datetime.utcnow() + timedelta(days=14)
|
||||||
|
|
||||||
|
exchange = Exchange(
|
||||||
|
slug="test123",
|
||||||
|
name="Test Exchange",
|
||||||
|
budget="$30",
|
||||||
|
max_participants=10,
|
||||||
|
registration_close_date=future_close_date,
|
||||||
|
exchange_date=future_exchange_date,
|
||||||
|
timezone="America/New_York",
|
||||||
|
state=Exchange.STATE_REGISTRATION_OPEN,
|
||||||
|
)
|
||||||
|
db.session.add(exchange)
|
||||||
|
db.session.flush()
|
||||||
|
|
||||||
|
participant = Participant(
|
||||||
|
exchange_id=exchange.id,
|
||||||
|
name="Test User",
|
||||||
|
email="test@example.com",
|
||||||
|
reminder_enabled=True,
|
||||||
|
)
|
||||||
|
db.session.add(participant)
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
# Make 3 requests (should succeed)
|
||||||
|
for _i in range(3):
|
||||||
|
response = client.post(
|
||||||
|
"/exchange/test123/request-access",
|
||||||
|
data={"email": "test@example.com"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# 4th request should be rate limited
|
||||||
|
response = client.post(
|
||||||
|
"/exchange/test123/request-access",
|
||||||
|
data={"email": "test@example.com"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 429 # Too Many Requests
|
||||||
|
|
||||||
|
def test_request_magic_link_validates_email(self, client, db):
|
||||||
|
"""Test that email format is validated."""
|
||||||
|
future_close_date = datetime.utcnow() + timedelta(days=7)
|
||||||
|
future_exchange_date = datetime.utcnow() + timedelta(days=14)
|
||||||
|
|
||||||
|
exchange = Exchange(
|
||||||
|
slug="test123",
|
||||||
|
name="Test Exchange",
|
||||||
|
budget="$30",
|
||||||
|
max_participants=10,
|
||||||
|
registration_close_date=future_close_date,
|
||||||
|
exchange_date=future_exchange_date,
|
||||||
|
timezone="America/New_York",
|
||||||
|
state=Exchange.STATE_REGISTRATION_OPEN,
|
||||||
|
)
|
||||||
|
db.session.add(exchange)
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/exchange/test123/request-access",
|
||||||
|
data={"email": "not-an-email"},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert b"valid email" in response.data
|
||||||
|
|
||||||
|
def test_request_access_page_displays(self, client, db):
|
||||||
|
"""Test that request access page displays correctly."""
|
||||||
|
future_close_date = datetime.utcnow() + timedelta(days=7)
|
||||||
|
future_exchange_date = datetime.utcnow() + timedelta(days=14)
|
||||||
|
|
||||||
|
exchange = Exchange(
|
||||||
|
slug="test123",
|
||||||
|
name="Test Exchange",
|
||||||
|
budget="$30",
|
||||||
|
max_participants=10,
|
||||||
|
registration_close_date=future_close_date,
|
||||||
|
exchange_date=future_exchange_date,
|
||||||
|
timezone="America/New_York",
|
||||||
|
state=Exchange.STATE_REGISTRATION_OPEN,
|
||||||
|
)
|
||||||
|
db.session.add(exchange)
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
response = client.get("/exchange/test123/request-access")
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert b"Test Exchange" in response.data
|
||||||
|
assert b'name="email"' in response.data
|
||||||
|
|
||||||
|
def test_invalid_exchange_slug_returns_404(self, client):
|
||||||
|
"""Test that invalid exchange slug returns 404."""
|
||||||
|
response = client.get("/exchange/invalid/request-access")
|
||||||
|
assert response.status_code == 404
|
||||||
Reference in New Issue
Block a user