Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/tests/security/test_kerberos.py: 0%

1# 

2# Licensed to the Apache Software Foundation (ASF) under one 

3# or more contributor license agreements. See the NOTICE file 

4# distributed with this work for additional information 

5# regarding copyright ownership. The ASF licenses this file 

6# to you under the Apache License, Version 2.0 (the 

7# "License"); you may not use this file except in compliance 

8# with the License. You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, 

13# software distributed under the License is distributed on an 

14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

15# KIND, either express or implied. See the License for the 

16# specific language governing permissions and limitations 

17# under the License. 

18from __future__ import annotations 

19 

20import logging 

21import shlex 

22from unittest import mock 

23 

24import pytest 

25 

26from airflow.security import kerberos 

27from airflow.security.kerberos import renew_from_kt 

28from tests.test_utils.config import conf_vars 

29 

30 

31class TestKerberos: 

32 @pytest.mark.parametrize( 

33 "kerberos_config, expected_cmd", 

34 [ 

35 ( 

36 {("kerberos", "reinit_frequency"): "42"}, 

37 [ 

38 "kinit", 

39 "-f", 

40 "-a", 

41 "-r", 

42 "42m", 

43 "-k", 

44 "-t", 

45 "keytab", 

46 "-c", 

47 "/tmp/airflow_krb5_ccache", 

48 "test-principal", 

49 ], 

50 ), 

51 ( 

52 {("kerberos", "forwardable"): "True", ("kerberos", "include_ip"): "True"}, 

53 [ 

54 "kinit", 

55 "-f", 

56 "-a", 

57 "-r", 

58 "3600m", 

59 "-k", 

60 "-t", 

61 "keytab", 

62 "-c", 

63 "/tmp/airflow_krb5_ccache", 

64 "test-principal", 

65 ], 

66 ), 

67 ( 

68 {("kerberos", "forwardable"): "False", ("kerberos", "include_ip"): "False"}, 

69 [ 

70 "kinit", 

71 "-F", 

72 "-A", 

73 "-r", 

74 "3600m", 

75 "-k", 

76 "-t", 

77 "keytab", 

78 "-c", 

79 "/tmp/airflow_krb5_ccache", 

80 "test-principal", 

81 ], 

82 ), 

83 ], 

84 ) 

85 @mock.patch("time.sleep", return_value=None) 

86 @mock.patch("airflow.security.kerberos.open", mock.mock_open(read_data=b"X-CACHECONF:")) 

87 @mock.patch("airflow.security.kerberos.NEED_KRB181_WORKAROUND", None) 

88 @mock.patch("airflow.security.kerberos.subprocess") 

89 def test_renew_from_kt(self, mock_subprocess, mock_sleep, kerberos_config, expected_cmd, caplog): 

90 expected_cmd_text = " ".join(shlex.quote(f) for f in expected_cmd) 

91 

92 with conf_vars(kerberos_config), caplog.at_level(logging.INFO, logger=kerberos.log.name): 

93 caplog.clear() 

94 mock_subprocess.Popen.return_value.__enter__.return_value.returncode = 0 

95 mock_subprocess.call.return_value = 0 

96 renew_from_kt(principal="test-principal", keytab="keytab") 

97 

98 assert caplog.messages == [ 

99 f"Re-initialising kerberos from keytab: {expected_cmd_text}", 

100 "Renewing kerberos ticket to work around kerberos 1.8.1: kinit -c /tmp/airflow_krb5_ccache -R", 

101 ] 

102 

103 assert mock_subprocess.Popen.call_args[0][0] == expected_cmd 

104 assert mock_subprocess.mock_calls == [ 

105 mock.call.Popen( 

106 expected_cmd, 

107 bufsize=-1, 

108 close_fds=True, 

109 stderr=mock_subprocess.PIPE, 

110 stdout=mock_subprocess.PIPE, 

111 universal_newlines=True, 

112 ), 

113 mock.call.Popen().__enter__(), 

114 mock.call.Popen().__enter__().wait(), 

115 mock.call.Popen().__exit__(None, None, None), 

116 mock.call.call(["kinit", "-c", "/tmp/airflow_krb5_ccache", "-R"], close_fds=True), 

117 ] 

118 

119 @mock.patch("airflow.security.kerberos.subprocess") 

120 @mock.patch("airflow.security.kerberos.NEED_KRB181_WORKAROUND", None) 

121 @mock.patch("airflow.security.kerberos.open", mock.mock_open(read_data=b"")) 

122 def test_renew_from_kt_without_workaround(self, mock_subprocess, caplog): 

123 mock_subprocess.Popen.return_value.__enter__.return_value.returncode = 0 

124 mock_subprocess.call.return_value = 0 

125 

126 with caplog.at_level(logging.INFO, logger=kerberos.log.name): 

127 caplog.clear() 

128 renew_from_kt(principal="test-principal", keytab="keytab") 

129 assert caplog.messages == [ 

130 "Re-initialising kerberos from keytab: " 

131 "kinit -f -a -r 3600m -k -t keytab -c /tmp/airflow_krb5_ccache test-principal" 

132 ] 

133 

134 assert mock_subprocess.mock_calls == [ 

135 mock.call.Popen( 

136 [ 

137 "kinit", 

138 "-f", 

139 "-a", 

140 "-r", 

141 "3600m", 

142 "-k", 

143 "-t", 

144 "keytab", 

145 "-c", 

146 "/tmp/airflow_krb5_ccache", 

147 "test-principal", 

148 ], 

149 bufsize=-1, 

150 close_fds=True, 

151 stderr=mock_subprocess.PIPE, 

152 stdout=mock_subprocess.PIPE, 

153 universal_newlines=True, 

154 ), 

155 mock.call.Popen().__enter__(), 

156 mock.call.Popen().__enter__().wait(), 

157 mock.call.Popen().__exit__(None, None, None), 

158 ] 

159 

160 @mock.patch("airflow.security.kerberos.subprocess") 

161 @mock.patch("airflow.security.kerberos.NEED_KRB181_WORKAROUND", None) 

162 def test_renew_from_kt_failed(self, mock_subprocess, caplog): 

163 mock_subp = mock_subprocess.Popen.return_value.__enter__.return_value 

164 mock_subp.returncode = 1 

165 mock_subp.stdout = mock.MagicMock(name="stdout", **{"readlines.return_value": ["STDOUT"]}) 

166 mock_subp.stderr = mock.MagicMock(name="stderr", **{"readlines.return_value": ["STDERR"]}) 

167 

168 with pytest.raises(SystemExit) as ctx: 

169 caplog.clear() 

170 renew_from_kt(principal="test-principal", keytab="keytab") 

171 assert ctx.value.code == 1 

172 

173 log_records = [record for record in caplog.record_tuples if record[0] == kerberos.log.name] 

174 assert len(log_records) == 2, log_records 

175 assert [lr[1] for lr in log_records] == [logging.INFO, logging.ERROR] 

176 assert [lr[2] for lr in log_records] == [ 

177 "Re-initialising kerberos from keytab: " 

178 "kinit -f -a -r 3600m -k -t keytab -c /tmp/airflow_krb5_ccache test-principal", 

179 "Couldn't reinit from keytab! `kinit' exited with 1.\nSTDOUT\nSTDERR", 

180 ] 

181 

182 assert mock_subprocess.mock_calls == [ 

183 mock.call.Popen( 

184 [ 

185 "kinit", 

186 "-f", 

187 "-a", 

188 "-r", 

189 "3600m", 

190 "-k", 

191 "-t", 

192 "keytab", 

193 "-c", 

194 "/tmp/airflow_krb5_ccache", 

195 "test-principal", 

196 ], 

197 bufsize=-1, 

198 close_fds=True, 

199 stderr=mock_subprocess.PIPE, 

200 stdout=mock_subprocess.PIPE, 

201 universal_newlines=True, 

202 ), 

203 mock.call.Popen().__enter__(), 

204 mock.call.Popen().__enter__().wait(), 

205 mock.call.Popen().__exit__(mock.ANY, mock.ANY, mock.ANY), 

206 ] 

207 

208 @mock.patch("airflow.security.kerberos.subprocess") 

209 @mock.patch("airflow.security.kerberos.NEED_KRB181_WORKAROUND", None) 

210 @mock.patch("airflow.security.kerberos.open", mock.mock_open(read_data=b"X-CACHECONF:")) 

211 @mock.patch("airflow.security.kerberos.get_hostname", return_value="HOST") 

212 @mock.patch("time.sleep", return_value=None) 

213 def test_renew_from_kt_failed_workaround(self, mock_sleep, mock_getfqdn, mock_subprocess, caplog): 

214 mock_subprocess.Popen.return_value.__enter__.return_value.returncode = 0 

215 mock_subprocess.call.return_value = 1 

216 

217 with pytest.raises(SystemExit) as ctx: 

218 caplog.clear() 

219 renew_from_kt(principal="test-principal", keytab="keytab") 

220 assert ctx.value.code == 1 

221 

222 log_records = [record for record in caplog.record_tuples if record[0] == kerberos.log.name] 

223 assert len(log_records) == 3, log_records 

224 assert [lr[1] for lr in log_records] == [logging.INFO, logging.INFO, logging.ERROR] 

225 assert [lr[2] for lr in log_records] == [ 

226 "Re-initialising kerberos from keytab: " 

227 "kinit -f -a -r 3600m -k -t keytab -c /tmp/airflow_krb5_ccache test-principal", 

228 "Renewing kerberos ticket to work around kerberos 1.8.1: kinit -c /tmp/airflow_krb5_ccache -R", 

229 "Couldn't renew kerberos ticket in order to work around " 

230 "Kerberos 1.8.1 issue. Please check that the ticket for 'test-principal/HOST' is still " 

231 "renewable:\n $ kinit -f -c /tmp/airflow_krb5_ccache\n" 

232 "If the 'renew until' date is the same as the 'valid starting' date, the ticket cannot be " 

233 "renewed. Please check your KDC configuration, and the ticket renewal policy (maxrenewlife) for " 

234 "the 'test-principal/HOST' and `krbtgt' principals.", 

235 ] 

236 

237 assert mock_subprocess.mock_calls == [ 

238 mock.call.Popen( 

239 [ 

240 "kinit", 

241 "-f", 

242 "-a", 

243 "-r", 

244 "3600m", 

245 "-k", 

246 "-t", 

247 "keytab", 

248 "-c", 

249 "/tmp/airflow_krb5_ccache", 

250 "test-principal", 

251 ], 

252 bufsize=-1, 

253 close_fds=True, 

254 stderr=mock_subprocess.PIPE, 

255 stdout=mock_subprocess.PIPE, 

256 universal_newlines=True, 

257 ), 

258 mock.call.Popen().__enter__(), 

259 mock.call.Popen().__enter__().wait(), 

260 mock.call.Popen().__exit__(None, None, None), 

261 mock.call.call(["kinit", "-c", "/tmp/airflow_krb5_ccache", "-R"], close_fds=True), 

262 ] 

263 

264 def test_run_without_keytab(self, caplog): 

265 with pytest.raises(SystemExit) as ctx: 

266 with caplog.at_level(logging.WARNING, logger=kerberos.log.name): 

267 caplog.clear() 

268 kerberos.run(principal="test-principal", keytab=None) 

269 assert ctx.value.code == 0 

270 assert caplog.messages == ["Keytab renewer not starting, no keytab configured"] 

271 

272 @mock.patch("airflow.security.kerberos.renew_from_kt") 

273 @mock.patch("time.sleep", return_value=None) 

274 def test_run(self, mock_sleep, mock_renew_from_kt): 

275 mock_renew_from_kt.side_effect = [1, 1, SystemExit(42)] 

276 with pytest.raises(SystemExit) as ctx: 

277 kerberos.run(principal="test-principal", keytab="/tmp/keytab") 

278 assert ctx.value.code == 42 

279 assert mock_renew_from_kt.mock_calls == [ 

280 mock.call("test-principal", "/tmp/keytab"), 

281 mock.call("test-principal", "/tmp/keytab"), 

282 mock.call("test-principal", "/tmp/keytab"), 

283 ]