forked from cartsnitch/cartsnitch
Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2f1833e90d | |||
| 5532b43e38 | |||
| 0be7ccd4b4 | |||
| 6d37cecdba | |||
| 3745f5be69 | |||
| abec954320 |
+2
-1
@@ -6,7 +6,7 @@ from logging.config import fileConfig
|
||||
from sqlalchemy import engine_from_config, pool
|
||||
|
||||
from alembic import context
|
||||
from cartsnitch_api.models.base import Base # noqa: F401 — imports all models for autogenerate
|
||||
from cartsnitch_api.models import Base # noqa: F401 — imports all models for autogenerate
|
||||
|
||||
config = context.config
|
||||
if config.config_file_name is not None:
|
||||
@@ -53,6 +53,7 @@ def run_migrations_online() -> None:
|
||||
# checkfirst=True ensures this is a no-op on existing databases.
|
||||
try:
|
||||
Base.metadata.create_all(bind=connection, checkfirst=True)
|
||||
connection.commit()
|
||||
except Exception as exc:
|
||||
import logging
|
||||
logging.getLogger("alembic.env").warning(
|
||||
|
||||
@@ -71,11 +71,14 @@ async def get_current_user(
|
||||
# 1. Check session cookie — prefer __Secure- variant (HTTPS) over plain (HTTP dev)
|
||||
cookie_token = request.cookies.get(SECURE_SESSION_COOKIE_NAME) or request.cookies.get(SESSION_COOKIE_NAME)
|
||||
if cookie_token:
|
||||
token = cookie_token
|
||||
# Better-Auth cookie format is "token.sessionId" — extract just the token part
|
||||
token = cookie_token.split(".")[0] if "." in cookie_token else cookie_token
|
||||
|
||||
# 2. Fall back to Bearer header
|
||||
if not token and credentials:
|
||||
token = credentials.credentials
|
||||
# Callers might pass the compound value here too
|
||||
raw = credentials.credentials
|
||||
token = raw.split(".")[0] if "." in raw else raw
|
||||
|
||||
if not token:
|
||||
raise HTTPException(
|
||||
|
||||
@@ -71,6 +71,56 @@ async def test_delete_me(client, auth_headers):
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_compound_cookie(client, db_engine):
|
||||
"""Compound cookie value (token.sessionId) must be parsed to extract the token part."""
|
||||
from tests.conftest import _create_test_user_and_session
|
||||
|
||||
_, session_token = await _create_test_user_and_session(
|
||||
client, db_engine, email="compound@example.com", display_name="Compound User"
|
||||
)
|
||||
compound = f"{session_token}.B0atkJCFxK1rZlwWPMK97nVO2LnyDun7"
|
||||
resp = await client.get(
|
||||
"/auth/me",
|
||||
headers={"Cookie": f"better-auth.session_token={compound}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["email"] == "compound@example.com"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_raw_token_cookie(client, db_engine):
|
||||
"""Raw token (no dot) in cookie must still work — regression guard."""
|
||||
from tests.conftest import _create_test_user_and_session
|
||||
|
||||
_, session_token = await _create_test_user_and_session(
|
||||
client, db_engine, email="rawcookie@example.com", display_name="Raw Cookie User"
|
||||
)
|
||||
resp = await client.get(
|
||||
"/auth/me",
|
||||
headers={"Cookie": f"better-auth.session_token={session_token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["email"] == "rawcookie@example.com"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_me_compound_bearer(client, db_engine):
|
||||
"""Compound Bearer token (token.sessionId) must be parsed to extract the token part."""
|
||||
from tests.conftest import _create_test_user_and_session
|
||||
|
||||
_, session_token = await _create_test_user_and_session(
|
||||
client, db_engine, email="compoundbearer@example.com", display_name="Compound Bearer User"
|
||||
)
|
||||
compound = f"{session_token}.B0atkJCFxK1rZlwWPMK97nVO2LnyDun7"
|
||||
resp = await client.get(
|
||||
"/auth/me",
|
||||
headers={"Authorization": f"Bearer {compound}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["email"] == "compoundbearer@example.com"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_expired_session_rejected(client, db_engine):
|
||||
"""Expired sessions must be rejected."""
|
||||
|
||||
Reference in New Issue
Block a user