Spaces:
Runtime error
Runtime error
File size: 9,730 Bytes
00a4363 1e1198f 00a4363 013ff92 b74c6ae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | from app.services.email import ConsoleEmailProvider
from tests.conftest import last_code
def _signup(client, email="writer@example.com", password="stardust-42"):
r = client.post("/api/auth/email/register/start", json={"email": email, "password": password, "name": "Ada L"})
assert r.status_code == 200, r.text
r = client.post("/api/auth/email/register/verify", json={"email": email, "code": last_code()})
assert r.status_code == 200, r.text
return r.json()
def test_signup_login_me_logout(client):
data = _signup(client)
assert data["created"] is True
assert any(e.subject == "Welcome to Notestack" for e in ConsoleEmailProvider.sent)
headers = {"Authorization": f"Bearer {data['access_token']}"}
assert client.get("/api/auth/me", headers=headers).json()["email"] == "writer@example.com"
r = client.post("/api/auth/email/login", json={"email": "WRITER@example.com", "password": "stardust-42"})
assert r.status_code == 200
assert client.post("/api/auth/logout", headers=headers).status_code == 200
assert client.get("/api/auth/me", headers=headers).status_code == 401 # token_version bumped
def test_wrong_code_and_attempt_limit(client):
client.post("/api/auth/email/register/start", json={"email": "a@b.co", "password": "longenough"})
for _ in range(5):
r = client.post("/api/auth/email/register/verify", json={"email": "a@b.co", "code": "000000"})
if last_code() == "000000":
return # astronomically unlikely collision
assert r.status_code == 400
r = client.post("/api/auth/email/register/verify", json={"email": "a@b.co", "code": last_code()})
assert r.json()["detail"]["code"] == "too_many_attempts"
def test_wrong_code_counts_down_and_lockout_allows_resend(client):
client.post("/api/auth/email/register/start", json={"email": "e@f.co", "password": "longenough"})
if last_code() == "000000":
return
r = client.post("/api/auth/email/register/verify", json={"email": "e@f.co", "code": "000000"})
assert r.json()["detail"]["message"].endswith("4 tries left.")
for _ in range(4):
r = client.post("/api/auth/email/register/verify", json={"email": "e@f.co", "code": "000000"})
assert r.json()["detail"]["code"] == "too_many_attempts"
# A dead code does not hold up a new one, and the pending sign up survives the lockout.
assert client.post("/api/auth/email/register/resend", json={"email": "e@f.co"}).status_code == 200
r = client.post("/api/auth/email/register/verify", json={"email": "e@f.co", "code": last_code()})
assert r.status_code == 200
def test_email_send_failure_is_reported(client, monkeypatch):
from app.services.email import email_service
monkeypatch.setattr(email_service, "send_verification_code", lambda to, code: False)
r = client.post("/api/auth/email/register/start", json={"email": "f@g.co", "password": "longenough"})
assert r.status_code == 502
assert r.json()["detail"]["code"] == "email_failed"
def test_forgot_password_unknown_and_google(client, db_session):
from app.services.auth_identity import resolve_or_create_google_user
r = client.post("/api/auth/password/forgot/start", json={"email": "nobody@example.com"})
assert r.status_code == 404
resolve_or_create_google_user(db_session, email="g2@example.com", google_id="456", name="G", avatar_url=None)
db_session.commit()
r = client.post("/api/auth/password/forgot/start", json={"email": "g2@example.com"})
assert r.status_code == 409
assert r.json()["detail"]["code"] == "wrong_provider"
def test_resend_cooldown(client):
client.post("/api/auth/email/register/start", json={"email": "c@d.co", "password": "longenough"})
r = client.post("/api/auth/email/register/resend", json={"email": "c@d.co"})
assert r.status_code == 429
def test_bad_password_and_weak_password(client):
_signup(client)
r = client.post("/api/auth/email/login", json={"email": "writer@example.com", "password": "nope-nope"})
assert r.status_code == 401
r = client.post("/api/auth/email/register/start", json={"email": "x@y.co", "password": "short"})
assert r.status_code == 400
def test_password_reset_revokes_old_tokens(client):
data = _signup(client)
old = {"Authorization": f"Bearer {data['access_token']}"}
client.post("/api/auth/password/forgot/start", json={"email": "writer@example.com"})
code = last_code()
r = client.post("/api/auth/password/forgot/check", json={"email": "writer@example.com", "code": code})
assert r.status_code == 200
r = client.post("/api/auth/password/forgot/complete",
json={"email": "writer@example.com", "code": code, "password": "new-orbit-99"})
assert r.status_code == 200
assert client.get("/api/auth/me", headers=old).status_code == 401
def test_google_user_cannot_use_password(client, db_session):
from app.services.auth_identity import resolve_or_create_google_user
resolve_or_create_google_user(db_session, email="g@example.com", google_id="123", name="G", avatar_url=None)
db_session.commit()
r = client.post("/api/auth/email/register/start", json={"email": "g@example.com", "password": "longenough"})
assert r.status_code == 409
assert r.json()["detail"]["code"] == "wrong_provider"
def test_delete_account(client):
data = _signup(client)
headers = {"Authorization": f"Bearer {data['access_token']}"}
assert client.post("/api/auth/delete-account", headers=headers).status_code == 200
r = client.post("/api/auth/email/login", json={"email": "writer@example.com", "password": "stardust-42"})
assert r.status_code == 403
def _google_redirect(client, monkeypatch, claims: dict, next_path="/app/notebooks"):
from urllib.parse import parse_qs, urlparse
from app.config import settings
from app.routers import auth as auth_router
monkeypatch.setattr(settings, "google_client_id", "cid")
monkeypatch.setattr(settings, "google_client_secret", "secret")
assert client.get("/api/auth/providers").json() == {"google": "redirect"}
r = client.get("/api/auth/google/start", params={"next": next_path}, follow_redirects=False)
assert r.status_code == 302
q = parse_qs(urlparse(r.headers["location"]).query)
assert q["redirect_uri"] == ["http://localhost:8000/api/auth/google/callback"]
monkeypatch.setattr(auth_router, "exchange_google_code", lambda code: {"nonce": q["nonce"][0], **claims})
r = client.get("/api/auth/google/callback", params={"code": "abc", "state": q["state"][0]},
follow_redirects=False)
assert r.status_code == 302
location = urlparse(r.headers["location"])
return location, parse_qs(location.fragment)
def test_google_redirect_flow(client, monkeypatch):
claims = {"email": "Star@Example.com", "email_verified": True, "sub": "g-1", "name": "Star"}
location, frag = _google_redirect(client, monkeypatch, claims)
assert location.path == "/auth/callback"
assert frag["next"] == ["/app/notebooks"]
r = client.post("/api/auth/ticket", json={"ticket": frag["ticket"][0]})
assert r.status_code == 200
data = r.json()
assert data["created"] is True and data["user"]["email"] == "star@example.com"
me = client.get("/api/auth/me", headers={"Authorization": f"Bearer {data['access_token']}"})
assert me.json()["auth_provider"] == "google"
# A ticket is not a session token.
assert client.get("/api/auth/me", headers={"Authorization": f"Bearer {frag['ticket'][0]}"}).status_code == 401
def test_google_redirect_rejects_open_redirect_and_email_accounts(client, monkeypatch):
_signup(client, email="taken@example.com")
claims = {"email": "taken@example.com", "email_verified": True, "sub": "g-2"}
location, frag = _google_redirect(client, monkeypatch, claims, next_path="//evil.example")
assert location.path == "/auth"
assert "password" in frag["error"][0]
def test_google_callback_requires_state_cookie(client, monkeypatch):
from datetime import timedelta
from app.auth import encode_signed
state = encode_signed({"typ": "oauth_state", "nonce": "n", "next": "/app"}, timedelta(minutes=5))
r = client.get("/api/auth/google/callback", params={"code": "abc", "state": state}, follow_redirects=False)
assert r.headers["location"].startswith("http://localhost:5173/auth#error=")
def test_safe_next():
from app.routers.auth import _safe_next
assert _safe_next("/app/notebooks/1?x=1") == "/app/notebooks/1?x=1"
for bad in (None, "", "https://evil.example", "//evil.example", r"/\evil.example"):
assert _safe_next(bad) == "/app"
def test_google_id_token_check_allows_for_a_laptop_clock_that_runs_behind(monkeypatch):
"""Seen on a dev machine whose clock was 2 seconds behind Google: google-auth refused the fresh ID token ("Token
used too early") and the writer only saw "Google sign in failed". Both Google flows now allow some drift."""
from app.config import settings
from app.routers import auth as auth_router
seen = []
class Reply:
def raise_for_status(self):
pass
def json(self):
return {"id_token": "t"}
def verify(token, request, audience=None, clock_skew_in_seconds=0):
seen.append(clock_skew_in_seconds)
return {"email": "a@b.co"}
monkeypatch.setattr(settings, "google_client_id", "cid")
monkeypatch.setattr(auth_router.httpx, "post", lambda *a, **k: Reply())
monkeypatch.setattr(auth_router.google_id_token, "verify_oauth2_token", verify)
auth_router.exchange_google_code("code")
assert seen == [auth_router.GOOGLE_CLOCK_SKEW_SECONDS] and seen[0] >= 5
|