Coverage for src/lexigram/admin/auth/session_manager.py: 12%

1"""Admin Session Manager — pure session-lifecycle business logic. 

2 

3All persistence is delegated to an injected ``SessionRepositoryProtocol``. This 

4module contains no SQL, no DDL, and no driver-specific code. 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import UTC, datetime, timedelta 

10from typing import Any 

11import uuid 

12 

13from lexigram.contracts.auth.fingerprint import generate_device_id 

14from lexigram.contracts.auth.models import UserSession 

15from lexigram.contracts.auth.repositories import SessionRepositoryProtocol 

16from lexigram.di.decorators import inject 

17from lexigram.logging import get_logger 

18 

19logger = get_logger(__name__) 

20 

21 

22@inject 

23class AdminSessionManager: 

24 """Lifecycle manager for admin user sessions. 

25 

26 Enforces concurrency limits, expiry, and activity tracking. All I/O is 

27 performed through the injected ``SessionRepositoryProtocol`` so the class remains 

28 fully testable without a real database. 

29 """ 

30 

31 DEFAULT_EXPIRY_DAYS = 30 

32 MAX_CONCURRENT_SESSIONS = 5 

33 

34 def __init__(self, repo: SessionRepositoryProtocol) -> None: 

35 """Initialise with the session repository. 

36 

37 Args: 

38 repo: Persistence abstraction for session records. Injected by 

39 the DI container; typically backed by 

40 ``AdminSessionSqlRepository``. 

41 """ 

42 self._repo = repo 

43 

44 async def create_session( 

45 self, 

46 user_id: str, 

47 fingerprint_data: dict[str, Any], 

48 ip_address: str | None = None, 

49 user_agent: str | None = None, 

50 expires_days: int = DEFAULT_EXPIRY_DAYS, 

51 ) -> UserSession: 

52 """Create and persist a new admin session. 

53 

54 Enforces ``MAX_CONCURRENT_SESSIONS`` by revoking the oldest session 

55 when the limit is reached before creating a new one. 

56 

57 Args: 

58 user_id: Admin user identifier. 

59 fingerprint_data: Device/browser fingerprint dict used to derive 

60 a stable ``device_id``. 

61 ip_address: Originating IP (optional, for audit metadata). 

62 user_agent: Request User-Agent header (optional). 

63 expires_days: Session lifetime in days. 

64 

65 Returns: 

66 Hydrated ``UserSession`` for the newly created session. 

67 """ 

68 device_id = generate_device_id(fingerprint_data) 

69 now = datetime.now(UTC) 

70 expires_at = now + timedelta(days=expires_days) 

71 

72 # Enforce concurrent-session cap — evict oldest first. 

73 active = await self.get_active_sessions(user_id) 

74 if len(active) >= self.MAX_CONCURRENT_SESSIONS: 

75 oldest = min(active, key=lambda s: s.last_active_at or s.created_at) # type: ignore[arg-type, return-value] 

76 await self.revoke_session(oldest.session_id) 

77 logger.info("Evicted oldest admin session for user %s", user_id) 

78 

79 session_id = str(uuid.uuid4()) 

80 payload: dict[str, Any] = { 

81 "session_id": session_id, 

82 "admin_id": user_id, 

83 "device_id": device_id, 

84 "ip_address": ip_address, 

85 "user_agent": user_agent, 

86 "fingerprint": fingerprint_data, 

87 "expires_at": expires_at, 

88 } 

89 

90 await self._repo.insert(payload) 

91 

92 session = UserSession( 

93 session_id=session_id, 

94 user_id=user_id, 

95 device_id=device_id, 

96 ip_address=ip_address or "", 

97 user_agent=user_agent or "", 

98 fingerprint=fingerprint_data, 

99 expires_at=expires_at, 

100 created_at=now, 

101 last_active_at=now, 

102 is_active=True, 

103 ) 

104 

105 logger.info("Created admin session %s for user %s", session_id, user_id) 

106 return session 

107 

108 async def validate_session(self, session_id: str) -> UserSession | None: 

109 """Validate a session and refresh its activity timestamp. 

110 

111 Args: 

112 session_id: Opaque session identifier to validate. 

113 

114 Returns: 

115 Hydrated ``UserSession`` if active and unexpired, else ``None``. 

116 """ 

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

118 if not row: 

119 return None 

120 

121 expires_at = row.get("expires_at") 

122 if expires_at and datetime.now(UTC) > expires_at: 

123 await self.revoke_session(session_id) 

124 logger.warning("Admin session %s expired", session_id) 

125 return None 

126 

127 await self.update_activity(session_id) 

128 return self._row_to_session(row) 

129 

130 async def get_session(self, session_id: str) -> UserSession | None: 

131 """Retrieve an admin session by ID without refreshing activity. 

132 

133 Args: 

134 session_id: Opaque session identifier. 

135 

136 Returns: 

137 Hydrated ``UserSession``, or ``None`` when absent or inactive. 

138 """ 

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

140 if not row: 

141 return None 

142 

143 expires_at = row.get("expires_at") 

144 if expires_at and datetime.now(UTC) > expires_at: 

145 await self.revoke_session(session_id) 

146 return None 

147 

148 return self._row_to_session(row) 

149 

150 async def get_active_sessions(self, user_id: str) -> list[UserSession]: 

151 """Return all active, non-expired sessions for an admin user. 

152 

153 Args: 

154 user_id: Admin owner identifier. 

155 

156 Returns: 

157 List of ``UserSession`` objects, newest-activity first. 

158 """ 

159 rows = await self._repo.find_active_by_user(user_id, datetime.now(UTC)) 

160 return [self._row_to_session(r) for r in rows] 

161 

162 async def revoke_session(self, session_id: str) -> bool: 

163 """Revoke a single admin session. 

164 

165 Args: 

166 session_id: Session to deactivate. 

167 

168 Returns: 

169 ``True`` after successful revocation. 

170 """ 

171 await self._repo.revoke(session_id) 

172 logger.info("Revoked admin session %s", session_id) 

173 return True 

174 

175 async def revoke_all_sessions(self, user_id: str) -> int: 

176 """Revoke every active session for an admin user. 

177 

178 Args: 

179 user_id: Owner whose sessions are to be revoked. 

180 

181 Returns: 

182 ``1`` to indicate the operation ran (count of affected sessions 

183 is not surfaced by all backends). 

184 """ 

185 await self._repo.revoke_all(user_id) 

186 logger.info("Revoked all admin sessions for user %s", user_id) 

187 return 1 

188 

189 async def update_activity(self, session_id: str) -> bool: 

190 """Refresh the last-active timestamp for a session. 

191 

192 Args: 

193 session_id: Session to touch. 

194 

195 Returns: 

196 ``True`` after the update is sent to the repository. 

197 """ 

198 await self._repo.update_activity(session_id, datetime.now(UTC)) 

199 return True 

200 

201 # ------------------------------------------------------------------ 

202 # Internal helpers 

203 # ------------------------------------------------------------------ 

204 

205 @staticmethod 

206 def _row_to_session(row: dict[str, Any]) -> UserSession: 

207 """Map a raw repository row dict to a ``UserSession`` value object.""" 

208 return UserSession( 

209 session_id=row["session_id"], 

210 user_id=row["admin_id"], 

211 device_id=row.get("device_id", ""), 

212 ip_address=row.get("ip_address", ""), 

213 user_agent=row.get("user_agent", ""), 

214 fingerprint=row.get("fingerprint", {}), 

215 expires_at=row.get("expires_at"), 

216 created_at=row.get("created_at"), 

217 last_active_at=row.get("last_active_at"), 

218 is_active=row.get("is_active", True), 

219 )