Coverage for custom_components/remote_logger/syslog/exporter.py: 98%

1from __future__ import annotations 

2 

3import asyncio 

4import contextlib 

5import logging 

6import os 

7import socket 

8import ssl 

9import time 

10from dataclasses import dataclass 

11from typing import TYPE_CHECKING, Any, cast 

12 

13from homeassistant.components.system_log import EVENT_SYSTEM_LOG 

14from homeassistant.const import CONF_HOST, CONF_PORT, CONF_PROTOCOL 

15 

16from custom_components.remote_logger.const import ( 

17 CONF_APP_NAME, 

18 CONF_BATCH_MAX_SIZE, 

19 CONF_CLIENT_TIMEOUT, 

20 CONF_FACILITY, 

21 CONF_SUPPRESS_SYSTEM_LOG_EVENT_NAME, 

22 CONF_USE_TLS, 

23 DEFAULT_CLIENT_TIMEOUT, 

24) 

25from custom_components.remote_logger.exporter import LogExporter, LogMessage, LogSubmission 

26from custom_components.remote_logger.helpers import flatten_event_data, isotimestamp 

27 

28from .const import ( 

29 DEFAULT_APP_NAME, 

30 DEFAULT_FACILITY, 

31 DEFAULT_SYSLOG_SEVERITY, 

32 PROTOCOL_UDP, 

33 SYSLOG_FACILITY_MAP, 

34 SYSLOG_SEVERITY_MAP, 

35) 

36 

37if TYPE_CHECKING: 

38 import datetime as dt 

39 from collections.abc import Mapping 

40 

41 from homeassistant.config_entries import ConfigEntry 

42 from homeassistant.core import HomeAssistant 

43 

44_LOGGER = logging.getLogger(__name__) 

45 

46 

47@dataclass 

48class SyslogMessage(LogMessage): 

49 payload: bytes 

50 

51 

52class SyslogSubmission(LogSubmission): 

53 def __init__(self, records: list[SyslogMessage], protocol: str) -> None: 

54 self.records: list[SyslogMessage] = records 

55 self.protocol: str = protocol 

56 

57 def for_display(self) -> dict[str, Any]: 

58 return { 

59 "protocol": self.protocol, 

60 "data": cast("str", os.linesep).join(r.payload.decode("utf-8") for r in self.records), 

61 } 

62 

63 

64class SyslogExporter(LogExporter): 

65 """Buffers system_log_event records and flushes them as RFC 5424 syslog messages.""" 

66 

67 logger_type = "syslog" 

68 

69 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: 

70 super().__init__(hass) 

71 self.name = entry.title 

72 self._in_progress: list[SyslogMessage] = [] 

73 self._lock = asyncio.Lock() 

74 

75 opts = {**entry.data, **entry.options} 

76 

77 self._host = opts[CONF_HOST] 

78 self._port = opts[CONF_PORT] 

79 self._protocol = opts.get(CONF_PROTOCOL, PROTOCOL_UDP) 

80 self.destination = (self._host, str(self._port), self._protocol) 

81 self._use_tls = opts.get(CONF_USE_TLS, False) 

82 self._app_name = opts.get(CONF_APP_NAME, DEFAULT_APP_NAME) 

83 facility_name = opts.get(CONF_FACILITY, DEFAULT_FACILITY) 

84 self._facility = SYSLOG_FACILITY_MAP.get(facility_name, 1) 

85 self._batch_max_size = opts.get(CONF_BATCH_MAX_SIZE, 10) 

86 self._client_timeout = opts.get(CONF_CLIENT_TIMEOUT, DEFAULT_CLIENT_TIMEOUT) 

87 self._suppress_system_log_event_name = opts.get(CONF_SUPPRESS_SYSTEM_LOG_EVENT_NAME, True) 

88 self._hostname = "-" 

89 

90 # TCP connection state (lazily created) 

91 self._tcp_reader: asyncio.StreamReader | None = None 

92 self._tcp_writer: asyncio.StreamWriter | None = None 

93 

94 # UDP transport state (lazily created) 

95 self._udp_transport: asyncio.DatagramTransport | None = None 

96 

97 self.endpoint_desc = ( 

98 f"syslog://{self._host}:{self._port} ({self._protocol.upper()}" 

99 f"{'+TLS' if self._use_tls and self._protocol != PROTOCOL_UDP else ''})" 

100 ) 

101 _LOGGER.info(f"remote_logger: syslog configured for {self.endpoint_desc}") 

102 

103 def create_log_record( 

104 self, 

105 event_data: Mapping[str, Any], 

106 event_type: str | None = None, 

107 time_fired: dt.datetime | None = None, 

108 message_override: list[str] | None = None, 

109 level_override: str | None = None, 

110 state_only: bool = False, 

111 ) -> SyslogMessage: 

112 """Convert a system_log_event payload to an RFC 5424 syslog message.""" 

113 """ 

114 "name": str 

115 "message": list(str) 

116 "level": str 

117 "source": (str,int) 

118 "timestamp": float 

119 "exception": str 

120 "count": int 

121 "first_occurred": float 

122 """ 

123 data = event_data 

124 level: str = level_override or data.get("level", "INFO").upper() 

125 severity = SYSLOG_SEVERITY_MAP.get(level, DEFAULT_SYSLOG_SEVERITY) 

126 pri = self._facility * 8 + severity 

127 

128 # RFC 3339 timestamp 

129 timestamp_s: float = data.get("timestamp", time_fired.timestamp() if time_fired else time.time()) 

130 timestamp = isotimestamp(timestamp_s) 

131 

132 # Message body 

133 messages: list[str] = message_override or data.get("message", []) 

134 msg = " ".join(messages) if messages else "-" 

135 

136 # Structured data with meta info 

137 sd = "-" 

138 sd_params: list[str] = [] 

139 if event_type == EVENT_SYSTEM_LOG: 

140 source = data.get("source") 

141 if source and isinstance(source, tuple): 

142 source_path, source_linenum = source 

143 sd_params.extend([f'code.file.path="{_sd_escape(source_path)}"', f'code.line.number="{source_linenum}"']) 

144 logger_name = data.get("name") 

145 if logger_name: 

146 sd_params.append(f'code.function.name="{_sd_escape(logger_name)}"') 

147 if data.get("count"): 

148 sd_params.append(f'exception.count="{data["count"]}"') 

149 if data.get("first_occurred"): 

150 sd_params.append(f'exception.first_occurred="{isotimestamp(data["first_occurred"])}"') 

151 

152 exception = data.get("exception") 

153 if exception: 

154 sd_params.append(f'exception.stacktrace="{data["exception"]}"') 

155 if not self._suppress_system_log_event_name: 

156 sd_params.append(f'eventName="{EVENT_SYSTEM_LOG}"') 

157 msgid: str = EVENT_SYSTEM_LOG 

158 else: 

159 msgid = "-" 

160 else: 

161 if event_type is not None: 

162 sd_params.append(f"eventName={event_type}") 

163 # Use HA event type as MSGID for non-system-log events; "-" otherwise 

164 msgid = event_type or "-" 

165 for k, v in data.items(): 

166 for flat_key, flat_val in flatten_event_data(f"event.data.{k}" if k != "event.data" else k, v, state_only): 

167 sd_params.append(f'{_sd_escape(flat_key)}="{_sd_escape(str(flat_val))}"') 

168 

169 if sd_params: 

170 sd = f"[opentelemetry {' '.join(sd_params)}]" 

171 

172 # RFC 5424: <PRI>VERSION SP TIMESTAMP SP HOSTNAME SP APP-NAME SP PROCID SP MSGID SP SD [SP MSG] 

173 # VERSION = 1, PROCID = - 

174 syslog_line = f"<{pri}>1 {timestamp} {self._hostname} {self._app_name} - {msgid} {sd} {msg}" 

175 

176 return SyslogMessage(payload=syslog_line.encode("utf-8", errors="replace")) 

177 

178 def log_direct(self, event_name: str, message: str, level: str, attributes: dict[str, Any] | None = None) -> None: 

179 """Buffer a custom syslog record without requiring a HA Event.""" 

180 severity = SYSLOG_SEVERITY_MAP.get(level.upper(), DEFAULT_SYSLOG_SEVERITY) 

181 pri = self._facility * 8 + severity 

182 timestamp = isotimestamp(time.time()) 

183 sd = "-" 

184 if attributes: 

185 sd_params = [f'{_sd_escape(k)}="{_sd_escape(str(v))}"' for k, v in attributes.items()] 

186 sd = f"[opentelemetry {' '.join(sd_params)}]" 

187 syslog_line = f"<{pri}>1 {timestamp} {self._hostname} {self._app_name} - {event_name} {sd} {message}" 

188 record = SyslogMessage(payload=syslog_line.encode("utf-8", errors="replace")) 

189 self._buffer.append(record) 

190 self.on_event() 

191 if len(self._buffer) >= self._batch_max_size: 

192 self._hass.async_create_task(self.flush()) 

193 

194 async def flush(self) -> None: 

195 """Flush all buffered log records to the syslog endpoint.""" 

196 records: list[SyslogMessage] | None = None 

197 async with self._lock: 

198 if not self._in_progress: 

199 if not self._buffer: 

200 return 

201 records = cast("list[SyslogMessage]", self._buffer.copy()) 

202 self._buffer.clear() 

203 

204 try: 

205 if records: 

206 self._in_progress = records 

207 else: 

208 self._in_progress = [m for m in self._in_progress if not m.sent] 

209 

210 if self._protocol == PROTOCOL_UDP: 

211 await self._send_udp(self._in_progress) 

212 else: 

213 await self._send_tcp(self._in_progress) 

214 sent: list[SyslogMessage] = [m for m in self._in_progress if m.sent] 

215 if sent: 

216 self.last_sent_payload = SyslogSubmission(sent, protocol=self._protocol) 

217 self.on_success() 

218 self._in_progress = [m for m in self._in_progress if not m.sent] 

219 except Exception as e: 

220 _LOGGER.exception("remote_logger: unexpected error sending syslog messages") 

221 self.on_posting_error(str(e)) 

222 

223 async def _send_udp(self, messages: list[SyslogMessage]) -> None: 

224 """Send syslog messages over UDP.""" 

225 try: 

226 if self._udp_transport is None or self._udp_transport.is_closing(): 

227 loop = asyncio.get_running_loop() 

228 self._udp_transport, _ = await loop.create_datagram_endpoint( 

229 asyncio.DatagramProtocol, 

230 remote_addr=(self._host, self._port), 

231 ) 

232 for msg in messages: 

233 self._udp_transport.sendto(msg.payload) 

234 msg.sent = True 

235 except OSError as err: 

236 _LOGGER.warning("remote_logger: failed to send syslog via UDP: %s", err) 

237 self._udp_transport = None 

238 

239 async def _send_tcp(self, messages: list[SyslogMessage]) -> None: 

240 """Send syslog messages over TCP with octet-counting framing (RFC 6587).""" 

241 try: 

242 if self._tcp_writer is None or self._tcp_writer.is_closing(): 

243 await self._connect_tcp() 

244 

245 writer: asyncio.StreamWriter | None = self._tcp_writer 

246 if writer is None: 

247 raise OSError("Unable to create TCP writer") # Set by _connect_tcp above 

248 

249 for msg in messages: 

250 # Octet-counting: "LEN SP MSG" 

251 frame = f"{len(msg.payload)} ".encode("ascii") + msg.payload 

252 writer.write(frame) 

253 await writer.drain() 

254 for msg in messages: 

255 msg.sent = True 

256 except (OSError, ConnectionError) as err: 

257 _LOGGER.warning("remote_logger: failed to send syslog via TCP: %s", err) 

258 await self._close_tcp() 

259 

260 async def _connect_tcp(self) -> None: 

261 """Establish a TCP connection to the syslog server.""" 

262 ssl_ctx: ssl.SSLContext | None = None 

263 if self._use_tls: 

264 ssl_ctx = ssl.create_default_context() 

265 

266 self._tcp_reader, self._tcp_writer = await asyncio.wait_for( 

267 asyncio.open_connection(self._host, self._port, ssl=ssl_ctx), 

268 timeout=self._client_timeout, 

269 ) 

270 

271 async def _close_tcp(self) -> None: 

272 """Close the TCP connection.""" 

273 if self._tcp_writer is not None: 

274 with contextlib.suppress(Exception): 

275 self._tcp_writer.close() 

276 await self._tcp_writer.wait_closed() 

277 self._tcp_writer = None 

278 self._tcp_reader = None 

279 

280 async def close(self) -> None: 

281 """Clean up transport resources.""" 

282 if self._udp_transport is not None: 

283 self._udp_transport.close() 

284 self._udp_transport = None 

285 await self._close_tcp() 

286 

287 

288def _sd_escape(value: str) -> str: 

289 """Escape special characters for RFC 5424 structured data values.""" 

290 return value.replace("\\", "\\\\").replace('"', '\\"').replace("]", "\\]") 

291 

292 

293async def validate(hass: Any, host: str, port: int, protocol: str, use_tls: bool) -> str | None: 

294 """Test connectivity to a syslog endpoint. Returns error key or None.""" 

295 loop = hass.loop 

296 try: 

297 if protocol == PROTOCOL_UDP: 

298 # Quick UDP test: just resolve and create a socket 

299 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 

300 try: 

301 sock.setblocking(False) 

302 await loop.run_in_executor(None, lambda: socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_DGRAM)) 

303 finally: 

304 sock.close() 

305 else: 

306 # TCP: actually connect 

307 ssl_ctx = True if use_tls else None 

308 _, writer = await asyncio.wait_for( 

309 asyncio.open_connection(host, port, ssl=ssl_ctx), 

310 timeout=10, 

311 ) 

312 writer.close() 

313 await writer.wait_closed() 

314 except (OSError, TimeoutError, ConnectionRefusedError) as err: 

315 _LOGGER.error("remote_logger: Syslog connect failed: %s", err) 

316 return "cannot_connect" 

317 except Exception as err: # ruff: ignore[blind-except] 

318 _LOGGER.error("remote_logger: Syslog connect unknown error: %s", err) 

319 return "unknown" 

320 return None