Coverage for src/lexigram/auth/session/cookie_backend.py: 100%

1"""Session-cookie authentication backend for SSR admin flows. 

2 

3This module provides :class:`SessionCookieBackend`, which authenticates 

4requests by reading a session-ID cookie, looking up the session in a 

5:class:`~lexigram.contracts.auth.repositories.SessionRepositoryProtocol` 

6implementation, and then resolving the owning user via a caller-supplied 

7async callable. 

8 

9This is intentionally separate from :class:`~lexigram.auth.authn.jwt.JWTTokenManager` 

10— it targets SSR pages where the browser (not a JS client) manages 

11credentials through an ``HttpOnly`` cookie rather than a ``Bearer`` header. 

12 

13Dependency graph (all injected, nothing imported from sibling extensions): 

14 

15 request → cookie → SessionRepositoryProtocol.find_active 

16 → user_fetcher(user_id) → AuthenticatedUserProtocol 

17""" 

18 

19from __future__ import annotations 

20 

21from collections.abc import Awaitable, Callable 

22from datetime import UTC, timedelta 

23from typing import Any 

24 

25from lexigram.contracts.auth.repositories import SessionRepositoryProtocol 

26from lexigram.contracts.auth.user import AuthenticatedUserProtocol 

27from lexigram.contracts.core.identity import IdGeneratorProtocol 

28from lexigram.logging import get_logger 

29from lexigram.primitives import clock as ambient_clock 

30 

31logger = get_logger(__name__) 

32 

33 

34class SessionCookieBackend: 

35 """Session-cookie authentication backend for SSR flows. 

36 

37 Reads session ID from cookie, resolves user from 

38 :class:`~lexigram.contracts.auth.repositories.SessionRepositoryProtocol`. 

39 

40 Because session records contain only a ``user_id``, resolving the full 

41 :class:`~lexigram.contracts.auth.user.AuthenticatedUserProtocol` requires 

42 a ``user_fetcher`` callable that looks up the concrete user object from 

43 whatever store the application uses (database, cache, etc.). This keeps 

44 the backend decoupled from any specific user-store implementation. 

45 

46 Constructor args: 

47 session_repository: Persistence backend for session records. 

48 user_fetcher: Async callable that accepts a ``user_id`` string and 

49 returns the matching :class:`AuthenticatedUserProtocol`, or 

50 ``None`` when no user exists for that ID. 

51 cookie_name: Name of the session cookie (default ``"session_id"``). 

52 secure: Whether to set the ``Secure`` flag (default ``True``). 

53 http_only: Whether to set the ``HttpOnly`` flag (default ``True``). 

54 same_site: The ``SameSite`` policy (default ``"lax"``). 

55 

56 Example:: 

57 

58 backend = SessionCookieBackend( 

59 session_repository=sql_session_repo, 

60 user_fetcher=user_service.get_authenticated_user, 

61 ) 

62 

63 # In an SSR handler: 

64 user = await backend.authenticate(request) 

65 if user is None: 

66 # redirect to /login 

67 ... 

68 """ 

69 

70 def __init__( 

71 self, 

72 session_repository: SessionRepositoryProtocol, 

73 user_fetcher: Callable[[str], Awaitable[AuthenticatedUserProtocol | None]], 

74 ids: IdGeneratorProtocol | None = None, 

75 cookie_name: str = "session_id", 

76 secure: bool = True, 

77 http_only: bool = True, 

78 same_site: str = "lax", 

79 ) -> None: 

80 from lexigram.identity.generator import Uuid4Generator 

81 

82 self._repo = session_repository 

83 self._user_fetcher = user_fetcher 

84 self._ids = ids if ids is not None else Uuid4Generator() 

85 self._cookie_name = cookie_name 

86 self._secure = secure 

87 self._http_only = http_only 

88 self._same_site = same_site 

89 

90 async def authenticate(self, request: Any) -> AuthenticatedUserProtocol | None: 

91 """Extract session cookie, validate, return user or ``None``. 

92 

93 Reads ``cookie_name`` from ``request.cookies``, fetches the 

94 corresponding active session record, resolves the user via 

95 ``user_fetcher``, and refreshes the ``last_active_at`` timestamp. 

96 

97 Args: 

98 request: The incoming request object. Must expose a 

99 ``cookies`` mapping attribute (Starlette / ASGI-compatible). 

100 

101 Returns: 

102 The authenticated user if the session is valid, ``None`` otherwise. 

103 """ 

104 session_id: str | None = request.cookies.get(self._cookie_name) 

105 if not session_id: 

106 return None 

107 

108 row = await self._repo.find_active(session_id) 

109 if row is None: 

110 logger.debug("session_not_found", session_id=session_id) 

111 return None 

112 

113 user_id: str = row["user_id"] 

114 user = await self._user_fetcher(user_id) 

115 if user is None: 

116 logger.warning( 

117 "session_user_not_found", 

118 session_id=session_id, 

119 user_id=user_id, 

120 ) 

121 return None 

122 

123 # Touch the last-active timestamp so the session stays warm. 

124 await self._repo.update_activity( 

125 session_id, ambient_clock.now().replace(tzinfo=UTC) 

126 ) 

127 

128 return user 

129 

130 async def login( 

131 self, 

132 response: Any, 

133 user_id: str, 

134 expires_in: int = 86400, 

135 ) -> str: 

136 """Create a session record and set the session cookie on *response*. 

137 

138 Args: 

139 response: The outgoing response object. Must expose a 

140 ``set_cookie`` method with ``key``, ``value``, ``max_age``, 

141 ``secure``, ``httponly``, and ``samesite`` kwargs 

142 (Starlette / ASGI-compatible). 

143 user_id: Identifier of the user to create a session for. 

144 expires_in: Session lifetime in **seconds** (default 86 400 = 1 day). 

145 

146 Returns: 

147 The newly created ``session_id``. 

148 """ 

149 session_id = self._ids.generate_for("Session") 

150 expires_at = ambient_clock.now().replace(tzinfo=UTC) + timedelta( 

151 seconds=expires_in 

152 ) 

153 

154 await self._repo.insert( 

155 { 

156 "session_id": session_id, 

157 "user_id": user_id, 

158 "device_id": "ssr", 

159 "expires_at": expires_at, 

160 } 

161 ) 

162 

163 response.set_cookie( 

164 key=self._cookie_name, 

165 value=session_id, 

166 max_age=expires_in, 

167 secure=self._secure, 

168 httponly=self._http_only, 

169 samesite=self._same_site, 

170 ) 

171 

172 logger.info("session_created", session_id=session_id, user_id=user_id) 

173 return session_id 

174 

175 async def logout(self, request: Any, response: Any) -> None: 

176 """Invalidate the session and clear the cookie. 

177 

178 Args: 

179 request: The incoming request. ``request.cookies`` is read to 

180 find the current session ID. 

181 response: The outgoing response. ``delete_cookie`` is called to 

182 remove the session cookie from the browser. 

183 """ 

184 session_id: str | None = request.cookies.get(self._cookie_name) 

185 if session_id: 

186 await self._repo.revoke(session_id) 

187 logger.info("session_revoked", session_id=session_id) 

188 

189 response.delete_cookie(key=self._cookie_name) 

190 

191 

192__all__ = ["SessionCookieBackend"]