Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,18 @@ Note that 0.3.3 was never released; 0.3.4 follows 0.3.2.
`CacheXError` if retries run out.
([#175](https://github.com/allen0099/FastAPI-CacheX/issues/175))

### Security

- **`rotate_session_id(request)` gives the request's session a new ID at
login.** With `FastAPICacheXSessionMiddleware`, a Starlette-style login that
only writes to `request.session` keeps the session ID the request arrived
with, so whoever planted that cookie was logged in too. Call it before
attaching the user; with no session loaded it does nothing, since the first
write starts a fresh one. The SESSION.md example used `SessionDep` and
answered `401` to new visitors; it now uses the helper, and the migration
section warns about the difference from Starlette.
([#225](https://github.com/allen0099/FastAPI-CacheX/issues/225))

## [0.3.7] - 2026-09-25

### Added
Expand Down
1 change: 1 addition & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ The library has four independent subsystems:
- Session token is passed via custom header (`X-Session-Token` by default) or `Authorization: Bearer` token.
- `SessionManagerProxy` mirrors the `BackendProxy` pattern for managing the `SessionManager` singleton.
- Key FastAPI dependencies: `get_session`, `require_session`, `get_optional_session` (in `session/dependencies.py`).
- `rotate_session_id(request)` (same module) regenerates the loaded session's ID at login against session fixation; a no-op when none was loaded. The middleware notices the changed ID and sends the new token.

**4. State Management (`fastapi_cachex/state/`)**
- `StateManager` provides one-time-use state tokens for OAuth flows. States are consumed (deleted) on first successful `consume_state()` call.
Expand Down
33 changes: 24 additions & 9 deletions docs/SESSION.md
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,11 @@ async def me(session=Depends(get_session)):
replacing `Session.data` with the dict's contents.
- Clearing it (`request.session.clear()`) on a session that had data deletes the backend session;
a cookie client also receives a `Set-Cookie` that expires the cookie.
- Logging in by writing to `request.session` keeps the session ID the request arrived with.
With Starlette's middleware the cookie *is* the session, so the login response replaces
whatever cookie was planted; here the cookie only names a server-side record, and a planted
one would be logged in along with the victim. Call `await rotate_session_id(request)` before
attaching the user (see [Regenerate the Session ID After Login](#5-regenerate-the-session-id-after-login)).
- Any access to `request.session` adds `Vary` for every request header read to find the token:
the headers checked in `token_source_priority` order (`header_name`, and `Authorization` when
bearer tokens are enabled) up to the one that carried the token. `Cookie` is added only when
Expand Down Expand Up @@ -605,25 +610,34 @@ match the bound one (or has no address) is treated as having no session.

### 5. Regenerate the Session ID After Login

Prevents session fixation attacks:
Prevents session fixation. The token a client arrives with may have been planted by
someone else (from a sibling subdomain, say); a login that keeps it hands that person a
logged-in session. Give the session a new ID before attaching the user:

```python
from fastapi_cachex.session.dependencies import SessionDep, SessionManagerDep
from fastapi_cachex.session import rotate_session_id


@app.post("/login")
async def login(request: Request, session: SessionDep, manager: SessionManagerDep):
async def login(request: Request):
... # verify the credentials
await manager.regenerate_session_id(session)
await rotate_session_id(request)
request.session["user_id"] = "123"
return {"ok": True}
```

`regenerate_session_id()` deletes the backend record under the old ID and saves the session under
a new ID, keeping its data, user, `created_at` and expiry. When the session is the request's own
(from `SessionDep`, `get_session` and friends), either middleware sees the new ID and sends a token
for it through the transport the request used: `Set-Cookie` for a cookie, the response header for a
header token. After that the old token no longer resolves to a session.
`rotate_session_id()` calls `SessionManager.regenerate_session_id()` on the request's
session, which deletes the backend record under the old ID and saves the session under a
new ID, keeping its data, user, `created_at` and expiry. Either middleware sees the new ID
and sends a token for it through the transport the request used: `Set-Cookie` for a
cookie, the response header for a header token. After that the old token no longer
resolves to a session. For a new visitor there is no session to rotate, so it returns
`False` and the first write starts a session under a fresh ID.

A handler that already holds the request's session object can call
`await manager.regenerate_session_id(session)` directly, with the same effect. Get it from
`get_optional_session` and skip the call when it is `None`; `SessionDep` answers `401` to a
visitor who has no session yet.

Outside a middleware, load the session with the same bindings the middleware would pass, and hand
the returned token to the client yourself:
Expand Down Expand Up @@ -659,6 +673,7 @@ from fastapi_cachex.session import (
get_optional_session, # optional authentication (None when there is no session)
require_session, # alias of get_session
get_session_manager, # the SessionManager registered by the middleware
rotate_session_id, # not a dependency: await it at login for a new session ID
)

# Type annotations
Expand Down
2 changes: 2 additions & 0 deletions fastapi_cachex/session/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from .dependencies import get_session_client_ip
from .dependencies import get_session_manager
from .dependencies import require_session
from .dependencies import rotate_session_id
from .manager import SessionManager
from .middleware import FastAPICacheXSessionMiddleware
from .middleware import SessionMiddleware
Expand All @@ -28,4 +29,5 @@
"get_session_client_ip",
"get_session_manager",
"require_session",
"rotate_session_id",
]
44 changes: 44 additions & 0 deletions fastapi_cachex/session/dependencies.py
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,50 @@ async def login(
return manager


async def rotate_session_id(request: Request) -> bool:
"""Give the request's session a new ID, as a defence against session fixation.

Call it at login, before attaching the user. A session token the client
arrived with may have been planted by someone else; after rotation the
old token no longer resolves, and the middleware sends the client a token
for the new ID through the transport the request used. The session keeps
its data, user and expiry.

With no session loaded there is nothing to rotate: the first write to
``request.session`` (or ``create_session()``) starts a session under a
fresh ID anyway. So this works for a new visitor and a returning one alike.

Example:
```python
from fastapi_cachex.session.dependencies import rotate_session_id


@app.post("/login")
async def login(request: Request):
... # verify the credentials
await rotate_session_id(request)
request.session["user_id"] = "123"
return {"ok": True}
```

Args:
request: FastAPI request object

Returns:
True if a loaded session was given a new ID, False if none was loaded

Raises:
HTTPException: 500 if no session middleware has registered a
SessionManager yet
"""
manager = get_session_manager(request)
session: Session | None = getattr(request.state, "__fastapi_cachex_session", None)
if session is None:
return False
await manager.regenerate_session_id(session)
return True


def get_session_client_ip(
request: Request,
manager: "SessionManager" = Depends(get_session_manager),
Expand Down
14 changes: 9 additions & 5 deletions i18n/zh-TW/docs/SESSION.md
Original file line number Diff line number Diff line change
Expand Up @@ -328,6 +328,7 @@ async def me(session=Depends(get_session)):
- 在沒有載入任何 Session 時寫入 `request.session`,會建立一個新的**匿名** Session(`SessionManager.create_anonymous_session()`,並依設定套用 IP / User-Agent 綁定),並透過該請求的傳輸方式傳回其權杖。
- 修改已載入 Session 的 `request.session`,會透過 `update_session()` 將新內容儲存到後端,以 dict 的內容取代 `Session.data`。
- 在原本有資料的 Session 上清除它(`request.session.clear()`),會刪除後端的 Session;Cookie 用戶端還會收到一個使 Cookie 過期的 `Set-Cookie`。
- 以寫入 `request.session` 的方式登入時,會沿用請求帶來的 Session ID。Starlette 的中介軟體中 Cookie *就是* Session,因此登入回應會取代任何被植入的 Cookie;這裡的 Cookie 只是指向伺服器端紀錄的名稱,被植入的 Cookie 會跟著受害者一起登入。請在附加使用者之前呼叫 `await rotate_session_id(request)`(見[登入後重新產生 Session ID](#5-regenerate-the-session-id-after-login))。
- 只要存取 `request.session`,就會為了尋找權杖而讀取過的每個請求標頭加入 `Vary`:依 `token_source_priority` 順序檢查的標頭(`header_name`,以及啟用 Bearer 權杖時的 `Authorization`),直到攜帶權杖的那一個為止。只有在沒有任何標頭攜帶權杖時才會讀取 Cookie,因此也只有這時才會加入 `Cookie`。

Cookie 一律為 `HttpOnly`;`Secure`、`SameSite`、`Domain`、`Path` 與 `Max-Age` 則依 `cookie_*` 設定。
Expand Down Expand Up @@ -507,21 +508,23 @@ config = SessionConfig(

### 5. 登入後重新產生 Session ID {#5-regenerate-the-session-id-after-login}

防止 Session 固定攻擊(session fixation):
防止 Session 固定攻擊(session fixation)。用戶端帶來的權杖可能是別人預先植入的(例如從同網域的其他子網域);若登入時沿用它,植入者就會拿到一個已登入的 Session。請在附加使用者之前為 Session 換一個新 ID:

```python
from fastapi_cachex.session.dependencies import SessionDep, SessionManagerDep
from fastapi_cachex.session import rotate_session_id


@app.post("/login")
async def login(request: Request, session: SessionDep, manager: SessionManagerDep):
async def login(request: Request):
... # 驗證帳號密碼
await manager.regenerate_session_id(session)
await rotate_session_id(request)
request.session["user_id"] = "123"
return {"ok": True}
```

`regenerate_session_id()` 會刪除舊 ID 底下的後端紀錄,並以新 ID 儲存該 Session,保留其資料、使用者、`created_at` 與過期時間。當該 Session 是請求本身的 Session(來自 `SessionDep`、`get_session` 等)時,任一個中介軟體都會看到新 ID,並透過該請求使用的傳輸方式送出對應的權杖:Cookie 使用 `Set-Cookie`,標頭權杖則使用回應標頭。之後舊的權杖就無法再解析出 Session。
`rotate_session_id()` 會對請求的 Session 呼叫 `SessionManager.regenerate_session_id()`,刪除舊 ID 底下的後端紀錄,並以新 ID 儲存該 Session,保留其資料、使用者、`created_at` 與過期時間。任一個中介軟體都會看到新 ID,並透過該請求使用的傳輸方式送出對應的權杖:Cookie 使用 `Set-Cookie`,標頭權杖則使用回應標頭。之後舊的權杖就無法再解析出 Session。新訪客沒有可換 ID 的 Session,因此它會回傳 `False`,第一次寫入時會以全新的 ID 建立 Session。

已經取得請求 Session 物件的 handler,也可以直接呼叫 `await manager.regenerate_session_id(session)`,效果相同。請從 `get_optional_session` 取得 Session,並在它為 `None` 時略過呼叫;`SessionDep` 會對還沒有 Session 的訪客回應 `401`。

在中介軟體之外,請以中介軟體會傳入的相同綁定值載入 Session,並自行將回傳的權杖交給用戶端:

Expand All @@ -546,6 +549,7 @@ from fastapi_cachex.session import (
get_optional_session, # 可選驗證(沒有 Session 時為 None)
require_session, # get_session 的別名
get_session_manager, # 中介軟體註冊的 SessionManager
rotate_session_id, # 不是依賴項:在登入時 await 它以取得新的 Session ID
)

# 型別註記
Expand Down
84 changes: 84 additions & 0 deletions tests/session/test_starlette_middleware.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from fastapi_cachex.backends.memory import MemoryBackend
from fastapi_cachex.session.config import SessionConfig
from fastapi_cachex.session.dependencies import get_session
from fastapi_cachex.session.dependencies import rotate_session_id
from fastapi_cachex.session.exceptions import SessionNotFoundError
from fastapi_cachex.session.manager import SessionManager
from fastapi_cachex.session.middleware import FastAPICacheXSessionMiddleware
Expand Down Expand Up @@ -838,6 +839,89 @@ async def test_deprecated_middleware_sends_regenerated_token(
await manager.get_session(old_token)


def _rotating_login_app(manager: SessionManager, config: SessionConfig) -> FastAPI:
"""An app with a Starlette-style login that rotates the ID first (#225)."""
app = FastAPI()
app.add_middleware(
FastAPICacheXSessionMiddleware, session_manager=manager, config=config
)

@app.post("/touch")
async def touch(request: Request):
request.session["cart"] = [1]
return {"ok": True}

@app.post("/login")
async def login(request: Request):
rotated = await rotate_session_id(request)
request.session["user_id"] = "alice"
return {"rotated": rotated}

return app


@pytest.mark.asyncio
async def test_rotate_session_id_defeats_a_planted_cookie(
manager: SessionManager, config: SessionConfig
) -> None:
"""A victim logging in with a planted cookie must not log the attacker in (#225).

Without rotation the login writes into the planted session and sends the
same token back, so the attacker's copy of the cookie now carries the
victim's user_id.
"""
app = _rotating_login_app(manager, config)
attacker = TestClient(app)
attacker.post("/touch")
planted = attacker.cookies[config.cookie_name]

victim = TestClient(app)
victim.cookies.set(config.cookie_name, planted)
response = victim.post("/login")

assert response.json() == {"rotated": True}
victim_token = _extract_cookie_token(
response.headers["set-cookie"], config.cookie_name
)
assert victim_token != planted
session, _ = await manager.get_session(victim_token)
assert session.data == {"cart": [1], "user_id": "alice"}
with pytest.raises(SessionNotFoundError):
await manager.get_session(planted)


def test_rotate_session_id_without_a_session(
manager: SessionManager, config: SessionConfig
) -> None:
"""A new visitor can log in: rotation is a no-op and the write starts a session."""
client = TestClient(_rotating_login_app(manager, config))

response = client.post("/login")

assert response.status_code == 200
assert response.json() == {"rotated": False}
assert config.cookie_name in client.cookies


@pytest.mark.asyncio
async def test_rotate_session_id_over_the_header(
manager: SessionManager, config: SessionConfig
) -> None:
"""A header client gets the rotated token back in the response header."""
_session, old_token = await manager.create_session(user=SessionUser(user_id="u"))
client = TestClient(_rotating_login_app(manager, config))

response = client.post("/login", headers={config.header_name: old_token})

assert response.json() == {"rotated": True}
assert "set-cookie" not in response.headers
new_token = response.headers[config.header_name]
session, _ = await manager.get_session(new_token)
assert session.data == {"user_id": "alice"}
with pytest.raises(SessionNotFoundError):
await manager.get_session(old_token)


def _vary(response: Any) -> set[str]:
"""The response's Vary header as a set of lowercased header names."""
return {
Expand Down
Loading