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._lock = threading.Lock() 

132 

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

134 """ 

135 Save a user to the in-memory storage. 

136 

137 Args: 

138 user: The user to save. 

139 

140 """ 

141 with self._lock: 

142 if any( 

143 (u.username == user.username or u.email == user.email) 

144 and u.id != user.id 

145 for u in self._storage.values() 

146 ): 

147 return Err( 

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

149 ) 

150 self._storage[user.id] = user 

151 return Ok(None) 

152 

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

154 """ 

155 Retrieve a user from the in-memory storage. 

156 

157 Args: 

158 user_id: The UUID of the user. 

159 

160 Returns: 

161 The UserInDB object if found, otherwise None. 

162 

163 """ 

164 with self._lock: 

165 return self._storage.get(user_id) 

166 

167 

168class UserService: 

169 """Service for managing users securely.""" 

170 

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

172 """ 

173 Initialize the UserService with a repository. 

174 

175 Args: 

176 user_repository: The repository to use for data access. 

177 

178 """ 

179 self._user_repository = user_repository 

180 

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

182 """ 

183 Create a new user. 

184 

185 Args: 

186 user_create: The user creation data. 

187 

188 Returns: 

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

190 

191 """ 

192 # Hash the password securely using the security module 

193 try: 

194 pwd_hash = hash_password(user_create.password) 

195 except ValueError as e: 

196 logger.warning( 

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

198 username=user_create.username, 

199 ) 

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

201 

202 user_id = uuid4() 

203 user_in_db = UserInDB( 

204 id=user_id, 

205 username=user_create.username, 

206 email=user_create.email, 

207 password_hash=pwd_hash, 

208 ) 

209 save_result = self._user_repository.save(user_in_db) 

210 match save_result: 

211 case Ok(): 

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

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

214 public_user = User( 

215 id=user_in_db.id, 

216 username=user_in_db.username, 

217 email=user_in_db.email, 

218 is_active=user_in_db.is_active, 

219 ) 

220 return Ok(public_user) 

221 case Err(error): 

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

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

224 case _: 

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

226 

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

228 """ 

229 Retrieve a user by ID using Result pattern. 

230 

231 Args: 

232 user_id: The UUID of the user. 

233 

234 Returns: 

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

236 

237 Example: 

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

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

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

241 ... else: 

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

243 

244 """ 

245 user_in_db = self._user_repository.get_by_id(user_id) 

246 if user_in_db is None: 

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

248 return Err(UserNotFoundError(user_id)) 

249 

250 public_user = User( 

251 id=user_in_db.id, 

252 username=user_in_db.username, 

253 email=user_in_db.email, 

254 is_active=user_in_db.is_active, 

255 ) 

256 return Ok(public_user)