Coverage for src/lexigram/auth/authn/apikeys.py: 96%

1"""API Key Management for Lexigram Auth.""" 

2 

3from __future__ import annotations 

4 

5from datetime import datetime, timedelta 

6import secrets 

7import string 

8 

9from lexigram.auth.authn.security import PasswordHasher 

10from lexigram.auth.models.apikey import APIKey 

11from lexigram.contracts.auth import APIKeyRepositoryProtocol 

12from lexigram.di.decorators import inject 

13from lexigram.logging import get_logger 

14 

15logger = get_logger(__name__) 

16 

17 

18@inject 

19class APIKeyManager: 

20 """Manages API keys for service-to-service authentication. 

21 

22 API keys follow the format: ``{prefix}_{random}`` — e.g. ``sk_live_abc123``. 

23 All persistence is delegated to an injected ``APIKeyRepositoryProtocol``; this 

24 class contains only key-management business logic. 

25 """ 

26 

27 KEY_LENGTH = 32 

28 DISPLAY_PREFIX_LENGTH = 8 

29 

30 def __init__(self, repo: APIKeyRepositoryProtocol) -> None: 

31 """Initialise with the API key repository. 

32 

33 Args: 

34 repo: Persistence abstraction for API key records. Injected by 

35 the DI container; typically backed by ``APIKeySqlRepository``. 

36 """ 

37 self._repo = repo 

38 

39 def generate_raw_key(self, prefix: str = "sk_live") -> str: 

40 """Generate a cryptographically secure random API key. 

41 

42 Args: 

43 prefix: Human-readable key family prefix (e.g. ``sk_live``). 

44 

45 Returns: 

46 Full raw key string in ``{prefix}_{random}`` format. 

47 """ 

48 alphabet = string.ascii_letters + string.digits 

49 key = "".join(secrets.choice(alphabet) for _ in range(self.KEY_LENGTH)) 

50 return f"{prefix}_{key}" 

51 

52 async def create_key( 

53 self, 

54 user_id: str, 

55 name: str, 

56 scopes: list[str] | None = None, 

57 expires_days: int | None = 365, 

58 prefix: str = "sk_live", 

59 ) -> tuple[str, APIKey]: 

60 """Create a new API key and persist its hash via the repository. 

61 

62 Args: 

63 user_id: Owner of the API key. 

64 name: Human-readable label for the key. 

65 scopes: Optional list of permission scopes. 

66 expires_days: Days until expiry (``None`` for non-expiring keys). 

67 prefix: Key family prefix written into the display prefix field. 

68 

69 Returns: 

70 ``(raw_key, api_key_object)`` — the raw key is only returned once. 

71 """ 

72 raw_key = self.generate_raw_key(prefix) 

73 key_hash = await PasswordHasher().hash(raw_key) 

74 display_prefix = raw_key[: self.DISPLAY_PREFIX_LENGTH] 

75 

76 expires_at: datetime | None = None 

77 if expires_days: 

78 expires_at = datetime.now() + timedelta(days=expires_days) 

79 

80 payload = { 

81 "name": name, 

82 "key_hash": key_hash, 

83 "prefix": display_prefix, 

84 "user_id": user_id, 

85 "scopes": scopes or [], 

86 "expires_at": expires_at, 

87 } 

88 

89 key_id = await self._repo.insert(payload) 

90 

91 api_key = APIKey( 

92 key_id=key_id, 

93 name=name, 

94 key_hash=key_hash, 

95 prefix=display_prefix, 

96 user_id=user_id, 

97 scopes=scopes or [], 

98 expires_at=expires_at, 

99 created_at=datetime.now(), 

100 updated_at=datetime.now(), 

101 ) 

102 

103 logger.info("Created API key for user %s: %s", user_id, name) 

104 return raw_key, api_key 

105 

106 async def validate_key( 

107 self, 

108 raw_key: str, 

109 ip_address: str | None = None, 

110 ) -> APIKey | None: 

111 """Validate a raw API key and refresh its last-used metadata. 

112 

113 Uses prefix pre-filtering for O(small) constant-time hash comparisons 

114 even with a large keyspace. 

115 

116 Args: 

117 raw_key: Full plain-text API key submitted by the caller. 

118 ip_address: Originating IP for audit metadata (optional). 

119 

120 Returns: 

121 Hydrated ``APIKey`` on success, ``None`` on invalid/expired key. 

122 """ 

123 display_prefix = raw_key[: self.DISPLAY_PREFIX_LENGTH] 

124 rows = await self._repo.find_by_prefix(display_prefix) 

125 

126 for row in rows: 

127 if await PasswordHasher().verify(raw_key, row["key_hash"]): 

128 expires_at = row.get("expires_at") 

129 if expires_at and expires_at < datetime.now(): 

130 logger.warning("Expired API key attempt: %s", row["id"]) 

131 continue 

132 

133 await self._repo.update_last_used(row["id"], ip_address) 

134 

135 return APIKey( 

136 key_id=str(row["id"]), 

137 name=row["name"], 

138 key_hash=row["key_hash"], 

139 prefix=row["prefix"], 

140 user_id=str(row["user_id"]), 

141 scopes=row["scopes"], 

142 expires_at=expires_at, 

143 last_used_at=datetime.now(), 

144 last_used_ip=ip_address, 

145 created_at=row["created_at"], 

146 updated_at=row["updated_at"], 

147 ) 

148 

149 return None 

150 

151 async def revoke_key(self, key_id: str) -> bool: 

152 """Revoke an API key permanently. 

153 

154 Args: 

155 key_id: Identifier of the key to revoke. 

156 

157 Returns: 

158 ``True`` when a live key was revoked; ``False`` when the id is 

159 unknown or already revoked. 

160 """ 

161 revoked = await self._repo.revoke(key_id) 

162 if revoked: 

163 logger.info("Revoked API key: %s", key_id) 

164 return revoked 

165 

166 async def list_keys(self, user_id: str) -> list[APIKey]: 

167 """List all active, non-revoked API keys for a user. 

168 

169 Args: 

170 user_id: Owner identifier. 

171 

172 Returns: 

173 List of hydrated ``APIKey`` objects. 

174 """ 

175 rows = await self._repo.find_by_user(user_id) 

176 return [ 

177 APIKey( 

178 key_id=str(row["id"]), 

179 name=row["name"], 

180 key_hash=row["key_hash"], 

181 prefix=row["prefix"], 

182 user_id=str(row["user_id"]), 

183 scopes=row["scopes"], 

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

185 last_used_at=row.get("last_used_at"), 

186 last_used_ip=row.get("last_used_ip"), 

187 created_at=row["created_at"], 

188 updated_at=row["updated_at"], 

189 ) 

190 for row in rows 

191 ] 

192 

193 

194__all__ = ["APIKeyManager"]