Coverage for custom_components/remote_logger/config_flow.py: 81%

1"""Config flow for the remote_logger integration.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import re 

7from typing import Any 

8 

9import voluptuous as vol 

10from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow 

11from homeassistant.const import CONF_HEADERS, CONF_HOST, CONF_PATH, CONF_PORT, CONF_PROTOCOL, CONF_TOKEN 

12from homeassistant.core import callback 

13from homeassistant.data_entry_flow import section 

14from homeassistant.helpers import selector 

15from homeassistant.helpers.aiohttp_client import async_get_clientsession 

16 

17from .const import ( 

18 BACKEND_OTEL, 

19 BACKEND_SYSLOG, 

20 CONF_BACKEND, 

21 CONF_CUSTOM_EVENTS, 

22 CONF_ENCODING, 

23 CONF_EVENT_BASED_LOGGING, 

24 CONF_LOG_HA_CORE_ACTIVITY, 

25 CONF_LOG_HA_CORE_CHANGES, 

26 CONF_LOG_HA_EVENT_BODY, 

27 CONF_LOG_HA_FULL_STATE_CHANGES, 

28 CONF_LOG_HA_LIFECYCLE, 

29 CONF_LOG_HA_STATE_CHANGES, 

30 CONF_LOG_LEVEL, 

31 CONF_RESOURCE_ATTRIBUTES, 

32 CONF_SUPPRESS_SYSTEM_LOG_EVENT_NAME, 

33 CONF_USE_TLS, 

34 DEFAULT_LOG_LEVEL, 

35 DOMAIN, 

36) 

37from .otel.const import ( 

38 CONF_TOKEN_TYPE, 

39 OTEL_DATA_SCHEMA, 

40 OTLP_LOGS_PATH, 

41 REAUTH_OTEL_DATA_SCHEMA, 

42 TOKEN_TYPE_BEARER, 

43 VALIDATION_MESSAGE, 

44) 

45from .otel.exporter import build_auth_header, parse_headers, parse_resource_attributes 

46from .otel.exporter import validate as otel_validate 

47from .syslog.const import SYSLOG_DATA_SCHEMA 

48from .syslog.exporter import validate as syslog_validate 

49 

50_LOGGER = logging.getLogger(__name__) 

51 

52 

53def _to_list(value: Any) -> list[str]: 

54 """Coerce a stored string (comma/newline-separated) or list to a list of strings.""" 

55 if isinstance(value, list): 

56 return value 

57 if not value: 

58 return [] 

59 return [v.strip() for v in re.split(r"[\n,]+", str(value)) if v.strip()] 

60 

61 

62COMMON_DATA_SCHEMA = vol.Schema({ 

63 vol.Optional(CONF_EVENT_BASED_LOGGING, default=False): selector.BooleanSelector(), 

64 vol.Optional(CONF_LOG_LEVEL, default=DEFAULT_LOG_LEVEL): selector.SelectSelector( 

65 selector.SelectSelectorConfig( 

66 options=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], 

67 mode=selector.SelectSelectorMode.DROPDOWN, 

68 ), 

69 ), 

70 vol.Required("ha_standard_events"): section( 

71 vol.Schema({ 

72 vol.Optional(CONF_LOG_HA_LIFECYCLE, default=False): selector.BooleanSelector(), 

73 vol.Optional(CONF_LOG_HA_CORE_CHANGES, default=False): selector.BooleanSelector(), 

74 vol.Optional(CONF_LOG_HA_CORE_ACTIVITY, default=False): selector.BooleanSelector(), 

75 vol.Optional(CONF_LOG_HA_STATE_CHANGES, default=False): selector.BooleanSelector(), 

76 vol.Optional(CONF_LOG_HA_FULL_STATE_CHANGES, default=False): selector.BooleanSelector(), 

77 }), 

78 {"collapsed": False}, 

79 ), 

80 vol.Optional(CONF_LOG_HA_EVENT_BODY, default=True): selector.BooleanSelector(), 

81 vol.Optional(CONF_SUPPRESS_SYSTEM_LOG_EVENT_NAME, default=True): selector.BooleanSelector(), 

82 vol.Optional(CONF_CUSTOM_EVENTS, default=[]): selector.TextSelector(selector.TextSelectorConfig(multiple=True)), 

83}) 

84 

85 

86def _build_endpoint_url(host: str, port: int, use_tls: bool, path: str = OTLP_LOGS_PATH) -> str: 

87 """Build the full OTLP endpoint URL.""" 

88 scheme = "https" if use_tls else "http" 

89 return f"{scheme}://{host}:{port}{path}" 

90 

91 

92class OtelLogsConfigFlow(ConfigFlow, domain=DOMAIN): 

93 """Handle a config flow for OpenTelemetry Log Exporter.""" 

94 

95 VERSION = 2 

96 

97 def __init__(self) -> None: 

98 super().__init__() 

99 self._pending_data: dict[str, Any] = {} 

100 

101 @staticmethod 

102 @callback 

103 def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow: 

104 """Return the options flow handler.""" 

105 return RemoteLoggerOptionsFlow(config_entry) 

106 

107 async def async_step_user( 

108 self, 

109 user_input: dict[str, Any] | None = None, 

110 ) -> ConfigFlowResult: 

111 """Show menu to choose backend type.""" 

112 return self.async_show_menu( 

113 step_id="user", 

114 menu_options=["otel", "syslog"], 

115 ) 

116 

117 async def async_step_otel(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

118 """Handle OpenTelemetry OTLP configuration.""" 

119 errors: dict[str, str] = {} 

120 

121 if user_input is not None: 

122 host = user_input[CONF_HOST] 

123 port = user_input[CONF_PORT] 

124 use_tls = user_input[CONF_USE_TLS] 

125 url = _build_endpoint_url(host, port, use_tls, user_input.get(CONF_PATH, OTLP_LOGS_PATH)) 

126 

127 # Validate header fields format before connecting 

128 extra_headers: dict[str, str] = {} 

129 token = user_input.get(CONF_TOKEN, "").strip() 

130 if token: 

131 if not use_tls: 

132 _LOGGER.warning("remote_logger: token configured without TLS; token will be sent in plain text") 

133 token_type = user_input.get(CONF_TOKEN_TYPE, TOKEN_TYPE_BEARER) 

134 extra_headers["Authorization"] = build_auth_header(token, token_type) 

135 raw_headers = "\n".join(user_input.get(CONF_HEADERS, [])) 

136 if raw_headers: 

137 try: 

138 extra_headers.update(parse_headers(raw_headers)) 

139 except ValueError: 

140 errors[CONF_HEADERS] = "invalid_headers" 

141 

142 # Validate connectivity 

143 if not errors: 

144 session = async_get_clientsession(self.hass, verify_ssl=use_tls) 

145 errors = await otel_validate(session, url, user_input[CONF_ENCODING], extra_headers or None) 

146 # Validate resource attributes format 

147 if not errors: 

148 raw_attrs = user_input.get(CONF_RESOURCE_ATTRIBUTES, "") 

149 if raw_attrs.strip(): 

150 try: 

151 parse_resource_attributes(raw_attrs) 

152 except ValueError: 

153 errors[CONF_RESOURCE_ATTRIBUTES] = "invalid_attributes" 

154 

155 if not errors: 

156 await self.async_set_unique_id(f"{DOMAIN}_{BACKEND_OTEL}_{host}_{port}") 

157 self._abort_if_unique_id_configured() 

158 self._pending_data = {**user_input, CONF_BACKEND: BACKEND_OTEL} 

159 self._pending_data["_title"] = f"OTLP @ {host}:{port}" 

160 return await self.async_step_common() 

161 

162 return self.async_show_form( 

163 step_id="otel", 

164 data_schema=self.add_suggested_values_to_schema(OTEL_DATA_SCHEMA, user_input or {}), 

165 errors=errors, 

166 description_placeholders={"validation_message": VALIDATION_MESSAGE}, 

167 ) 

168 

169 async def async_step_reauth(self, entry_data: dict[str, Any]) -> ConfigFlowResult: 

170 """Initiate reauth after authentication failure.""" 

171 return await self.async_step_reauth_otel() 

172 

173 async def async_step_reauth_otel(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

174 """Handle re-entry of the bearer token after an authentication failure.""" 

175 reauth_entry = self._get_reauth_entry() 

176 errors: dict[str, str] = {} 

177 

178 if user_input is not None: 

179 token = user_input.get(CONF_TOKEN, "").strip() 

180 extra_headers: dict[str, str] = {} 

181 if token: 

182 token_type = user_input.get(CONF_TOKEN_TYPE, TOKEN_TYPE_BEARER) 

183 extra_headers["Authorization"] = build_auth_header(token, token_type) 

184 raw_headers = "\n".join(reauth_entry.data.get(CONF_HEADERS, [])) 

185 if raw_headers: 

186 extra_headers.update(parse_headers(raw_headers)) 

187 

188 url = _build_endpoint_url( 

189 reauth_entry.data[CONF_HOST], 

190 reauth_entry.data[CONF_PORT], 

191 reauth_entry.data[CONF_USE_TLS], 

192 reauth_entry.data.get(CONF_PATH, OTLP_LOGS_PATH), 

193 ) 

194 session = async_get_clientsession(self.hass, verify_ssl=reauth_entry.data[CONF_USE_TLS]) 

195 errors = await otel_validate(session, url, reauth_entry.data[CONF_ENCODING], extra_headers or None) 

196 if not errors: 

197 return self.async_update_reload_and_abort( 

198 reauth_entry, 

199 data_updates={ 

200 CONF_TOKEN: user_input[CONF_TOKEN], 

201 CONF_TOKEN_TYPE: user_input.get(CONF_TOKEN_TYPE, TOKEN_TYPE_BEARER), 

202 }, 

203 ) 

204 

205 return self.async_show_form( 

206 step_id="reauth_otel", 

207 data_schema=self.add_suggested_values_to_schema(REAUTH_OTEL_DATA_SCHEMA, user_input or {}), 

208 errors=errors, 

209 description_placeholders={"validation_message": VALIDATION_MESSAGE}, 

210 ) 

211 

212 async def async_step_syslog(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

213 """Handle Syslog RFC 5424 configuration.""" 

214 errors: dict[str, str] = {} 

215 

216 if user_input is not None: 

217 host = user_input[CONF_HOST] 

218 port = user_input[CONF_PORT] 

219 protocol = user_input[CONF_PROTOCOL] 

220 use_tls = user_input.get(CONF_USE_TLS, False) 

221 

222 # Validate connectivity 

223 error = await syslog_validate(self.hass, host, port, protocol, use_tls) 

224 if error: 

225 errors["base"] = error 

226 

227 if not errors: 

228 await self.async_set_unique_id(f"{DOMAIN}_{BACKEND_SYSLOG}") 

229 self._abort_if_unique_id_configured() 

230 self._pending_data = {**user_input, CONF_BACKEND: BACKEND_SYSLOG} 

231 self._pending_data["_title"] = f"Syslog @ {host}:{port} ({protocol.upper()})" 

232 return await self.async_step_common() 

233 

234 return self.async_show_form( 

235 step_id="syslog", 

236 data_schema=self.add_suggested_values_to_schema(SYSLOG_DATA_SCHEMA, user_input or {}), 

237 errors=errors, 

238 ) 

239 

240 async def async_step_common(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

241 """Configure common event subscription options.""" 

242 errors: dict[str, str] = {} 

243 if user_input is not None: 

244 flat: dict[str, Any] = {k: v for k, v in user_input.items() if not isinstance(v, dict)} 

245 for v in user_input.values(): 

246 if isinstance(v, dict): 

247 flat.update(v) 

248 if flat.get(CONF_LOG_HA_STATE_CHANGES) and flat.get(CONF_LOG_HA_FULL_STATE_CHANGES): 

249 errors[CONF_LOG_HA_FULL_STATE_CHANGES] = "state_changes_exclusive" 

250 else: 

251 title = self._pending_data.pop("_title") 

252 return self.async_create_entry( 

253 title=title, 

254 data={**self._pending_data, **flat}, 

255 ) 

256 

257 return self.async_show_form( 

258 step_id="common", 

259 data_schema=self.add_suggested_values_to_schema(COMMON_DATA_SCHEMA, user_input or {}), 

260 errors=errors, 

261 description_placeholders={ 

262 "learn_more": "[Learn about HA events](https://www.home-assistant.io/docs/configuration/events/)", 

263 }, 

264 ) 

265 

266 

267class RemoteLoggerOptionsFlow(OptionsFlow): 

268 """Allow editing connection details and event subscriptions after setup.""" 

269 

270 def __init__(self, config_entry: ConfigEntry) -> None: 

271 self._config_entry = config_entry 

272 self._pending_options: dict[str, Any] = {} 

273 

274 async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

275 """Dispatch to the backend-specific connection form.""" 

276 backend = self._config_entry.data.get(CONF_BACKEND, BACKEND_OTEL) 

277 if backend == BACKEND_SYSLOG: 

278 return await self.async_step_syslog() 

279 return await self.async_step_otel() 

280 

281 async def async_step_otel(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

282 """Handle OTLP connection settings.""" 

283 errors: dict[str, str] = {} 

284 merged = {**self._config_entry.data, **self._config_entry.options} 

285 

286 if user_input is not None: 

287 host = user_input[CONF_HOST] 

288 port = user_input[CONF_PORT] 

289 use_tls = user_input[CONF_USE_TLS] 

290 url = _build_endpoint_url(host, port, use_tls, user_input.get(CONF_PATH, OTLP_LOGS_PATH)) 

291 

292 extra_headers: dict[str, str] = {} 

293 token = user_input.get(CONF_TOKEN, "").strip() 

294 if token: 

295 token_type = user_input.get(CONF_TOKEN_TYPE, TOKEN_TYPE_BEARER) 

296 extra_headers["Authorization"] = build_auth_header(token, token_type) 

297 raw_headers = "\n".join(user_input.get(CONF_HEADERS, [])) 

298 if raw_headers: 

299 try: 

300 extra_headers.update(parse_headers(raw_headers)) 

301 except ValueError: 

302 errors[CONF_HEADERS] = "invalid_headers" 

303 

304 if not errors: 

305 session = async_get_clientsession(self.hass, verify_ssl=use_tls) 

306 errors = await otel_validate(session, url, user_input[CONF_ENCODING], extra_headers or None) 

307 

308 if not errors: 

309 raw_attrs = user_input.get(CONF_RESOURCE_ATTRIBUTES, "") 

310 if raw_attrs.strip(): 

311 try: 

312 parse_resource_attributes(raw_attrs) 

313 except ValueError: 

314 errors[CONF_RESOURCE_ATTRIBUTES] = "invalid_attributes" 

315 

316 if not errors: 

317 self._pending_options = user_input 

318 return await self.async_step_events() 

319 

320 suggested = user_input or {**merged, CONF_HEADERS: _to_list(merged.get(CONF_HEADERS, []))} 

321 return self.async_show_form( 

322 step_id="otel", 

323 data_schema=self.add_suggested_values_to_schema(OTEL_DATA_SCHEMA, suggested), 

324 errors=errors, 

325 description_placeholders={"validation_message": VALIDATION_MESSAGE}, 

326 ) 

327 

328 async def async_step_syslog(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

329 """Handle Syslog connection settings.""" 

330 errors: dict[str, str] = {} 

331 merged = {**self._config_entry.data, **self._config_entry.options} 

332 

333 if user_input is not None: 

334 host = user_input[CONF_HOST] 

335 port = user_input[CONF_PORT] 

336 protocol = user_input[CONF_PROTOCOL] 

337 use_tls = user_input.get(CONF_USE_TLS, False) 

338 

339 error = await syslog_validate(self.hass, host, port, protocol, use_tls) 

340 if error: 

341 errors["base"] = error 

342 

343 if not errors: 

344 self._pending_options = user_input 

345 return await self.async_step_events() 

346 

347 return self.async_show_form( 

348 step_id="syslog", 

349 data_schema=self.add_suggested_values_to_schema(SYSLOG_DATA_SCHEMA, user_input or merged), 

350 errors=errors, 

351 ) 

352 

353 async def async_step_events(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: 

354 """Handle event subscription options.""" 

355 errors: dict[str, str] = {} 

356 if user_input is not None: 

357 flat: dict[str, Any] = {k: v for k, v in user_input.items() if not isinstance(v, dict)} 

358 for v in user_input.values(): 

359 if isinstance(v, dict): 

360 flat.update(v) 

361 if flat.get(CONF_LOG_HA_STATE_CHANGES) and flat.get(CONF_LOG_HA_FULL_STATE_CHANGES): 

362 errors[CONF_LOG_HA_FULL_STATE_CHANGES] = "state_changes_exclusive" 

363 else: 

364 return self.async_create_entry(title="", data={**self._pending_options, **flat}) 

365 

366 if user_input is not None: 

367 suggested = user_input 

368 else: 

369 merged = {**self._config_entry.data, **self._config_entry.options} 

370 suggested = { 

371 CONF_EVENT_BASED_LOGGING: merged.get(CONF_EVENT_BASED_LOGGING, False), 

372 CONF_LOG_LEVEL: merged.get(CONF_LOG_LEVEL, DEFAULT_LOG_LEVEL), 

373 "ha_standard_events": { 

374 CONF_LOG_HA_LIFECYCLE: merged.get(CONF_LOG_HA_LIFECYCLE, False), 

375 CONF_LOG_HA_CORE_CHANGES: merged.get(CONF_LOG_HA_CORE_CHANGES, False), 

376 CONF_LOG_HA_CORE_ACTIVITY: merged.get(CONF_LOG_HA_CORE_ACTIVITY, False), 

377 CONF_LOG_HA_STATE_CHANGES: merged.get(CONF_LOG_HA_STATE_CHANGES, False), 

378 CONF_LOG_HA_FULL_STATE_CHANGES: merged.get(CONF_LOG_HA_FULL_STATE_CHANGES, False), 

379 }, 

380 CONF_LOG_HA_EVENT_BODY: merged.get(CONF_LOG_HA_EVENT_BODY, False), 

381 CONF_SUPPRESS_SYSTEM_LOG_EVENT_NAME: merged.get(CONF_SUPPRESS_SYSTEM_LOG_EVENT_NAME, True), 

382 CONF_CUSTOM_EVENTS: _to_list(merged.get(CONF_CUSTOM_EVENTS, [])), 

383 } 

384 return self.async_show_form( 

385 step_id="events", 

386 data_schema=self.add_suggested_values_to_schema(COMMON_DATA_SCHEMA, suggested), 

387 errors=errors, 

388 description_placeholders={ 

389 "learn_more": "[Learn about HA events](https://www.home-assistant.io/docs/configuration/events/)", 

390 }, 

391 )