Coverage for src/app/secure_system.py: 100%

1""" 

2Secure System Module. 

3 

4This module demonstrates a secure implementation of a user management service 

5following strict typing and security guidelines. 

6""" 

7 

8import threading 

9from abc import ABC, abstractmethod 

10from uuid import UUID, uuid4 

11 

12from pydantic import EmailStr, Field, SecretStr 

13from pydantic.networks import IPvAnyAddress 

14 

15from taipanstack.core.result import Err, Ok, Result 

16from taipanstack.security import SecureBaseModel, hash_password 

17from taipanstack.utils.logging import get_logger 

18 

19# Configure logger 

20logger = get_logger(__name__) 

21 

22 

23class UserNotFoundError(Exception): 

24 """Exception raised when a user is not found.""" 

25 

26 def __init__(self, user_id: UUID) -> None: 

27 """Initialize the exception with the user ID.""" 

28 self.user_id = user_id 

29 super().__init__(f"User with ID {user_id} not found.") 

30 

31 

32class UserAlreadyExistsError(Exception): 

33 """Exception raised when a user already exists.""" 

34 

35 def __init__(self, message: str) -> None: 

36 """Initialize the exception with a message.""" 

37 self.message = message 

38 super().__init__(message) 

39 

40 

41class UserCreationError(Exception): 

42 """Exception class for user creation errors.""" 

43 

44 def __init__(self, message: str = "Failed to create user") -> None: 

45 """Initialize the exception with a message.""" 

46 self.message = message 

47 super().__init__(message) 

48 

49 

50class UserCreate(SecureBaseModel): 

51 """ 

52 Model for creating a new user. 

53 

54 Attributes: 

55 username: The username of the user. 

56 email: The email address of the user. 

57 password: The password of the user (will be treated as a secret). 

58 ip_address: The IP address from which the user is registering. 

59 

60 """ 

61 

62 username: str = Field(..., min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$") 

63 email: EmailStr 

64 password: SecretStr 

65 ip_address: IPvAnyAddress | None = None 

66 

67 

68class User(SecureBaseModel): 

69 """ 

70 Model representing a registered user. 

71 

72 Attributes: 

73 id: Unique identifier for the user. 

74 username: The username of the user. 

75 email: The email address of the user. 

76 is_active: Whether the user account is active. 

77 

78 """ 

79 

80 id: UUID 

81 username: str 

82 email: EmailStr 

83 is_active: bool = True 

84 

85 

86class UserInDB(User): 

87 """ 

88 Model representing a registered user in the database. 

89 

90 Attributes: 

91 password_hash: The hashed password of the user. 

92 

93 """ 

94 

95 password_hash: str 

96 

97 

98class UserRepository(ABC): 

99 """Abstract base class for user data access.""" 

100 

101 @abstractmethod 

102 def save(self, user: UserInDB) -> Result[None, UserAlreadyExistsError]: 

103 """ 

104 Save a user to the repository. 

105 

106 Args: 

107 user: The user to save. 

108 

109 """ 

110 

111 @abstractmethod 

112 def get_by_id(self, user_id: UUID) -> UserInDB | None: 

113 """ 

114 Retrieve a user by their ID. 

115 

116 Args: 

117 user_id: The UUID of the user. 

118 

119 Returns: 

120 The UserInDB object if found, otherwise None. 

121 

122 """ 

123 

124 

125class InMemoryUserRepository(UserRepository): 

126 """In-memory implementation of UserRepository.""" 

127 

128 def __init__(self) -> None: 

129 """Initialize the in-memory repository.""" 

130 self._storage: dict[UUID, UserInDB] = {} 

131 self._emails: set[str] = set() 

132 self._usernames: set[str] = set() 

133 self._lock = threading.Lock() 

134 

135 def save(self, user: UserInDB) -> Result[None, UserAlreadyExistsError]: 

136 """ 

137 Save a user to the in-memory storage. 

138 

139 Args: 

140 user: The user to save. 

141 

142 """ 

143 with self._lock: 

144 existing = self._storage.get(user.id) 

145 if existing is None: 

146 if user.email in self._emails or user.username in self._usernames: 

147 return Err( 

148 UserAlreadyExistsError(f"User {user.username} already exists.") 

149 ) 

150 else: 

151 email_conflict = ( 

152 user.email != existing.email and user.email in self._emails 

153 ) 

154 user_conflict = ( 

155 user.username != existing.username 

156 and user.username in self._usernames 

157 ) 

158 if email_conflict or user_conflict: 

159 return Err( 

160 UserAlreadyExistsError(f"User {user.username} already exists.") 

161 ) 

162 self._emails.discard(existing.email) 

163 self._usernames.discard(existing.username) 

164 

165 self._storage[user.id] = user 

166 self._emails.add(user.email) 

167 self._usernames.add(user.username) 

168 return Ok(None) 

169 

170 def get_by_id(self, user_id: UUID) -> UserInDB | None: 

171 """ 

172 Retrieve a user from the in-memory storage. 

173 

174 Args: 

175 user_id: The UUID of the user. 

176 

177 Returns: 

178 The UserInDB object if found, otherwise None. 

179 

180 """ 

181 with self._lock: 

182 return self._storage.get(user_id) 

183 

184 

185class UserService: 

186 """Service for managing users securely.""" 

187 

188 def __init__(self, user_repository: UserRepository) -> None: 

189 """ 

190 Initialize the UserService with a repository. 

191 

192 Args: 

193 user_repository: The repository to use for data access. 

194 

195 """ 

196 self._user_repository = user_repository 

197 

198 def create_user(self, user_create: UserCreate) -> Result[User, UserCreationError]: 

199 """ 

200 Create a new user. 

201 

202 Args: 

203 user_create: The user creation data. 

204 

205 Returns: 

206 Ok(User) on success, Err(UserCreationError) on failure. 

207 

208 """ 

209 # Hash the password securely using the security module 

210 try: 

211 pwd_hash = hash_password(user_create.password) 

212 except ValueError as e: 

213 logger.warning( 

214 "Failed to create user (invalid password)", 

215 username=user_create.username, 

216 ) 

217 return Err(UserCreationError(message=str(e))) 

218 

219 user_id = uuid4() 

220 user_in_db = UserInDB( 

221 id=user_id, 

222 username=user_create.username, 

223 email=user_create.email, 

224 password_hash=pwd_hash, 

225 ) 

226 save_result = self._user_repository.save(user_in_db) 

227 match save_result: 

228 case Ok(): 

229 logger.info("User created successfully", user_id=user_in_db.id) 

230 # Return the public User model, excluding the password hash 

231 public_user = User( 

232 id=user_in_db.id, 

233 username=user_in_db.username, 

234 email=user_in_db.email, 

235 is_active=user_in_db.is_active, 

236 ) 

237 return Ok(public_user) 

238 case Err(error): 

239 logger.warning("Failed to create user", user_id=user_in_db.id) 

240 return Err(UserCreationError(message=str(error))) 

241 case _: 

242 return Err(UserCreationError(message="Unknown save error")) # type: ignore[unreachable] 

243 

244 def get_user(self, user_id: UUID) -> Result[User, UserNotFoundError]: 

245 """ 

246 Retrieve a user by ID using Result pattern. 

247 

248 Args: 

249 user_id: The UUID of the user. 

250 

251 Returns: 

252 Ok(User) if found, Err(UserNotFoundError) if not found. 

253 

254 Example: 

255 >>> result = service.get_user(some_id) 

256 >>> if isinstance(result, Ok): 

257 ... print(f"Found: {result.unwrap().username}") 

258 ... else: 

259 ... print(f"Not found: {result.unwrap_err().user_id}") 

260 

261 """ 

262 user_in_db = self._user_repository.get_by_id(user_id) 

263 if user_in_db is None: 

264 logger.warning("User lookup failed", user_id=user_id) 

265 return Err(UserNotFoundError(user_id)) 

266 

267 public_user = User( 

268 id=user_in_db.id, 

269 username=user_in_db.username, 

270 email=user_in_db.email, 

271 is_active=user_in_db.is_active, 

272 ) 

273 return Ok(public_user)