Coverage for app/backend/src/tests/test_sanitized_bytes.py: 100%

1from google.protobuf import empty_pb2 

2 

3from couchers.interceptors import _sanitized_bytes 

4from couchers.proto import api_pb2, auth_pb2, conversations_pb2 

5 

6 

7class TestSanitizedBytes: 

8 """Test suite for _sanitized_bytes function.""" 

9 

10 def test_none_input(self): 

11 """Test that None input returns None.""" 

12 result = _sanitized_bytes(None) 

13 assert result is None 

14 

15 def test_empty_message(self): 

16 """Test that an empty message is serialized correctly.""" 

17 proto = empty_pb2.Empty() 

18 result = _sanitized_bytes(proto) 

19 

20 assert isinstance(result, bytes) 

21 # Verify we can deserialize it back 

22 deserialized = empty_pb2.Empty.FromString(result) 

23 assert deserialized == proto 

24 

25 def test_message_with_no_sensitive_fields(self): 

26 """Test that messages without sensitive fields are not modified.""" 

27 # AuthRes has no sensitive fields 

28 proto = auth_pb2.AuthRes(user_id=12345, jailed=False) 

29 result = _sanitized_bytes(proto) 

30 

31 deserialized = auth_pb2.AuthRes.FromString(result) 

32 assert deserialized.user_id == 12345 

33 assert deserialized.jailed is False 

34 

35 def test_message_with_sensitive_field(self): 

36 """Test that sensitive fields are cleared from messages.""" 

37 # AuthReq has a sensitive password field 

38 proto = auth_pb2.AuthReq(user="testuser", password="supersecret123", remember_device=True) 

39 # the plaintext is present in the wire bytes unsanitized, but must not survive sanitization 

40 assert b"supersecret123" in proto.SerializeToString() 

41 result = _sanitized_bytes(proto) 

42 assert b"supersecret123" not in result 

43 

44 deserialized = auth_pb2.AuthReq.FromString(result) 

45 

46 # Non-sensitive fields should remain 

47 assert deserialized.user == "testuser" 

48 assert deserialized.remember_device is True 

49 

50 # Sensitive field should be cleared 

51 assert deserialized.password == "" 

52 

53 def test_message_with_nested_message_non_repeated(self): 

54 """Test that nested messages are recursively sanitized (non-repeated case).""" 

55 # SignupFlowReq contains a nested SignupAccount message 

56 proto = auth_pb2.SignupFlowReq( 

57 flow_token="token123", 

58 account=auth_pb2.SignupAccount(username="nesteduser", password="nestedsecret", city="Boston"), 

59 ) 

60 assert b"nestedsecret" in proto.SerializeToString() 

61 result = _sanitized_bytes(proto) 

62 assert b"nestedsecret" not in result 

63 

64 deserialized = auth_pb2.SignupFlowReq.FromString(result) 

65 

66 # Top-level fields should remain 

67 assert deserialized.flow_token == "token123" 

68 

69 # Nested message non-sensitive fields should remain 

70 assert deserialized.account.username == "nesteduser" 

71 assert deserialized.account.city == "Boston" 

72 

73 # Nested message sensitive field should be cleared 

74 assert deserialized.account.password == "" 

75 

76 def test_message_with_empty_nested_message(self): 

77 """Test that empty nested message fields don't cause errors (continue branch).""" 

78 # Create a message where nested message field is default/empty 

79 # SignupFlowRes with auth_res not set (optional field) 

80 proto = auth_pb2.SignupFlowRes( 

81 flow_token="token456", 

82 need_basic=True, 

83 need_account=False, 

84 # auth_res is not set 

85 ) 

86 result = _sanitized_bytes(proto) 

87 

88 deserialized = auth_pb2.SignupFlowRes.FromString(result) 

89 assert deserialized.flow_token == "token456" 

90 assert deserialized.need_basic is True 

91 assert deserialized.need_account is False 

92 # auth_res should still be empty/default 

93 assert not deserialized.HasField("auth_res") 

94 

95 def test_message_with_empty_repeated_field(self): 

96 """ 

97 Test that empty repeated message fields trigger the continue branch. 

98 

99 This covers line 171-172 where submessage evaluates to False (empty list). 

100 """ 

101 # Create a message with a repeated field but don't add any items 

102 proto = conversations_pb2.GetGroupChatMessagesRes( 

103 last_message_id=999, 

104 no_more=True, 

105 # messages field is empty (default empty repeated field) 

106 ) 

107 result = _sanitized_bytes(proto) 

108 

109 deserialized = conversations_pb2.GetGroupChatMessagesRes.FromString(result) 

110 assert deserialized.last_message_id == 999 

111 assert deserialized.no_more is True 

112 # Repeated field should be empty 

113 assert len(deserialized.messages) == 0 

114 

115 def test_message_with_repeated_messages(self): 

116 """Test that repeated message fields are recursively sanitized.""" 

117 # Create a message with repeated nested messages 

118 # Using GetGroupChatMessagesRes which has repeated Message fields 

119 proto = conversations_pb2.GetGroupChatMessagesRes(last_message_id=100, no_more=False) 

120 

121 # Add messages to the repeated field 

122 msg1 = proto.messages.add() 

123 msg1.message_id = 1 

124 msg1.author_user_id = 123 

125 msg1.text.text = "Hello" 

126 

127 msg2 = proto.messages.add() 

128 msg2.message_id = 2 

129 msg2.author_user_id = 456 

130 msg2.text.text = "World" 

131 

132 result = _sanitized_bytes(proto) 

133 

134 deserialized = conversations_pb2.GetGroupChatMessagesRes.FromString(result) 

135 

136 # Check that repeated messages are preserved 

137 assert len(deserialized.messages) == 2 

138 assert deserialized.messages[0].message_id == 1 

139 assert deserialized.messages[0].text.text == "Hello" 

140 assert deserialized.messages[0].author_user_id == 123 

141 assert deserialized.messages[1].message_id == 2 

142 assert deserialized.messages[1].text.text == "World" 

143 assert deserialized.messages[1].author_user_id == 456 

144 

145 def test_deeply_nested_messages(self): 

146 """Test that deeply nested messages are recursively sanitized.""" 

147 # SignupFlowRes contains AuthRes which is nested 

148 proto = auth_pb2.SignupFlowRes( 

149 flow_token="deep_token", 

150 auth_res=auth_pb2.AuthRes(user_id=789, jailed=False), 

151 need_basic=False, 

152 need_account=True, 

153 ) 

154 result = _sanitized_bytes(proto) 

155 

156 deserialized = auth_pb2.SignupFlowRes.FromString(result) 

157 

158 # All nested data should be preserved (no sensitive fields in this structure) 

159 assert deserialized.flow_token == "deep_token" 

160 assert deserialized.auth_res.user_id == 789 

161 assert deserialized.auth_res.jailed is False 

162 assert deserialized.need_basic is False 

163 assert deserialized.need_account is True 

164 

165 def test_multiple_nested_messages_with_sensitive_fields(self): 

166 """ 

167 Test sanitization when a message has multiple nested messages with sensitive fields. 

168 

169 This ensures the function properly handles multiple different nested message fields. 

170 """ 

171 # Create a SignupFlowReq with multiple nested messages 

172 proto = auth_pb2.SignupFlowReq( 

173 flow_token="repeat_token", 

174 basic=auth_pb2.SignupBasic(name="Test User", email="test@example.com"), 

175 account=auth_pb2.SignupAccount(username="testuser", password="shouldberemoved"), 

176 ) 

177 assert b"shouldberemoved" in proto.SerializeToString() 

178 result = _sanitized_bytes(proto) 

179 assert b"shouldberemoved" not in result 

180 

181 deserialized = auth_pb2.SignupFlowReq.FromString(result) 

182 

183 # Check basic nested message 

184 assert deserialized.basic.name == "Test User" 

185 assert deserialized.basic.email == "test@example.com" 

186 

187 # Check account nested message with sensitive field 

188 assert deserialized.account.username == "testuser" 

189 assert deserialized.account.password == "" 

190 

191 def test_message_preserves_original(self): 

192 """Test that the original message is not modified (deepcopy is used).""" 

193 original = auth_pb2.AuthReq(user="original_user", password="original_password") 

194 

195 # Store original values 

196 original_user = original.user 

197 original_password = original.password 

198 

199 assert b"original_password" in original.SerializeToString() 

200 # Call _sanitized_bytes 

201 result = _sanitized_bytes(original) 

202 assert b"original_password" not in result 

203 

204 # Original should not be modified 

205 assert original.user == original_user 

206 assert original.password == original_password 

207 

208 # But the result should have password cleared 

209 deserialized = auth_pb2.AuthReq.FromString(result) 

210 assert deserialized.user == original_user 

211 assert deserialized.password == "" 

212 

213 def test_message_with_non_message_type_fields(self): 

214 """Test messages with primitive fields (no message_type).""" 

215 # Create a message with only primitive types 

216 proto = auth_pb2.AuthRes(user_id=12345, jailed=True) 

217 result = _sanitized_bytes(proto) 

218 

219 deserialized = auth_pb2.AuthRes.FromString(result) 

220 assert deserialized.user_id == 12345 

221 assert deserialized.jailed is True 

222 

223 def test_complex_nested_structure(self): 

224 """Test a complex nested structure to ensure all branches are covered.""" 

225 # Create a complex nested message 

226 proto = auth_pb2.SignupFlowReq( 

227 flow_token="complex_token", 

228 basic=auth_pb2.SignupBasic(name="Complex User", email="complex@example.com", invite_code="INVITE123"), 

229 account=auth_pb2.SignupAccount( 

230 username="complexuser", 

231 password="complexsecret", 

232 city="Seattle", 

233 lat=47.6062, 

234 lng=-122.3321, 

235 birthdate="1985-05-15", 

236 gender="other", 

237 hosting_status=api_pb2.HOSTING_STATUS_CAN_HOST, 

238 accept_tos=True, 

239 ), 

240 feedback=auth_pb2.ContributorForm(ideas="Great platform!", contribute=auth_pb2.CONTRIBUTE_OPTION_YES), 

241 email_token="email_verification_token", 

242 ) 

243 

244 assert b"complexsecret" in proto.SerializeToString() 

245 result = _sanitized_bytes(proto) 

246 assert b"complexsecret" not in result 

247 

248 deserialized = auth_pb2.SignupFlowReq.FromString(result) 

249 

250 # Verify all non-sensitive fields are preserved 

251 assert deserialized.flow_token == "complex_token" 

252 assert deserialized.basic.name == "Complex User" 

253 assert deserialized.basic.email == "complex@example.com" 

254 assert deserialized.basic.invite_code == "INVITE123" 

255 assert deserialized.account.username == "complexuser" 

256 assert deserialized.account.city == "Seattle" 

257 assert deserialized.account.lat == 47.6062 

258 assert deserialized.account.birthdate == "1985-05-15" 

259 assert deserialized.feedback.ideas == "Great platform!" 

260 assert deserialized.email_token == "email_verification_token" 

261 

262 # Verify sensitive field is cleared 

263 assert deserialized.account.password == ""