Coverage for app/security_alerts.py: 100%

1"""Real-time alerting for escalated [SECURITY] log events. 

2 

3Attaches a SecurityAlertHandler to the 'openhangar' logger. When a WARNING+ 

4record whose message contains '[SECURITY]' matches one of the escalated event 

5types, it fires up to three delivery channels — ntfy, email, webhook — each 

6gated by its own env var. Channels that are not configured are silently skipped. 

7Delivery failures are logged and never re-raised; alerting must not break the app. 

8 

9Env vars (all optional): 

10 OPENHANGAR_ALERT_NTFY_TOPIC_URL — ntfy topic URL (hosted or self-hosted) 

11 OPENHANGAR_ALERT_NTFY_TOKEN — ntfy access token, for a self-hosted 

12 instance with auth-default-access 

13 other than "allow" (also accepts 

14 OPENHANGAR_ALERT_NTFY_TOKEN_FILE) 

15 OPENHANGAR_ALERT_EMAIL_TO — recipient address for alert emails 

16 OPENHANGAR_ALERT_WEBHOOK_URL — generic HTTP POST endpoint (Slack, etc.) 

17 

18Email alerts reuse the existing OPENHANGAR_SMTP_* env vars. 

19""" 

20 

21import json 

22import logging 

23import os 

24import smtplib 

25import threading 

26import time 

27import urllib.error 

28import urllib.request 

29from email.mime.text import MIMEText 

30 

31from init import _env_or_file # pyright: ignore[reportMissingImports] 

32 

33_ESCALATED: frozenset[str] = frozenset( 

34 { 

35 "auth.login.account_locked", 

36 "auth.login.account_blocked", 

37 "auth.totp.replay", 

38 # Disabling 2FA is a classic account-takeover step (an attacker locking 

39 # out the legitimate owner); rare enough to alert on without noise. 

40 "auth.totp.disabled", 

41 "users.role.changed", 

42 "users.access.revoked", 

43 # An unverifiable backup is only discovered at the worst possible 

44 # moment (mid-disaster) unless it's alerted on immediately. 

45 "backup.verification_failed", 

46 } 

47) 

48 

49_DEBOUNCE_SECONDS = 60 

50 

51_log = logging.getLogger(__name__) 

52 

53_DEFAULT_FORMATTER = logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s") 

54 

55 

56class SecurityAlertHandler(logging.Handler): 

57 """Logging handler that fires real-time alerts for escalated security events.""" 

58 

59 def __init__(self) -> None: 

60 super().__init__(level=logging.WARNING) 

61 self.setFormatter(_DEFAULT_FORMATTER) 

62 self._debounce: dict[str, float] = {} 

63 self._lock = threading.Lock() 

64 

65 def emit(self, record: logging.LogRecord) -> None: 

66 if record.levelno < self.level: 

67 return 

68 try: 

69 raw = record.getMessage() 

70 if "[SECURITY]" not in raw: 

71 return 

72 

73 parts = raw.split() 

74 try: 

75 sec_idx = parts.index("[SECURITY]") 

76 except ValueError: 

77 return 

78 if sec_idx + 1 >= len(parts): 

79 return 

80 

81 event_type = parts[sec_idx + 1] 

82 if event_type not in _ESCALATED: 

83 return 

84 

85 now = time.monotonic() 

86 with self._lock: 

87 if now - self._debounce.get(event_type, 0.0) < _DEBOUNCE_SECONDS: 

88 return 

89 self._debounce[event_type] = now 

90 

91 body = self.format(record) 

92 self._dispatch(event_type, body) 

93 except Exception: # noqa: BLE001 -- logging.Handler.emit() contract: never raise, call handleError 

94 self.handleError(record) 

95 

96 def _dispatch(self, event_type: str, detail: str) -> None: 

97 ntfy_url = os.environ.get("OPENHANGAR_ALERT_NTFY_TOPIC_URL", "").strip() 

98 ntfy_token = _env_or_file("ALERT_NTFY_TOKEN") 

99 alert_email = os.environ.get("OPENHANGAR_ALERT_EMAIL_TO", "").strip() 

100 webhook_url = os.environ.get("OPENHANGAR_ALERT_WEBHOOK_URL", "").strip() 

101 

102 if ntfy_url: 

103 self._send_ntfy(ntfy_url, ntfy_token, event_type, detail) 

104 if alert_email: 

105 self._send_email(alert_email, event_type, detail) 

106 if webhook_url: 

107 self._send_webhook(webhook_url, event_type, detail) 

108 

109 def _send_ntfy(self, url: str, token: str, event_type: str, detail: str) -> None: 

110 try: 

111 headers = { 

112 "Title": f"OpenHangar security alert: {event_type}", 

113 "Priority": "high", 

114 "Tags": "warning,lock", 

115 } 

116 if token: 

117 headers["Authorization"] = f"Bearer {token}" 

118 req = urllib.request.Request( 

119 url, 

120 data=detail.encode("utf-8"), 

121 headers=headers, 

122 method="POST", 

123 ) 

124 with urllib.request.urlopen(req, timeout=10): # nosec B310 # scheme restricted to http(s) at startup (OPENHANGAR_ALERT_NTFY_TOPIC_URL check in init.py) 

125 pass 

126 except Exception as exc: # noqa: BLE001 -- one alert channel's failure must not affect others 

127 _log.error("Security alert: ntfy delivery failed: %s", exc) 

128 

129 def _send_email(self, to: str, event_type: str, detail: str) -> None: 

130 try: 

131 host = os.environ.get("OPENHANGAR_SMTP_HOST", "").strip() 

132 from_addr = os.environ.get("OPENHANGAR_SMTP_FROM_ADDRESS", "").strip() 

133 if not host or not from_addr: 

134 _log.error( 

135 "Security alert: email delivery skipped — " 

136 "OPENHANGAR_SMTP_HOST or OPENHANGAR_SMTP_FROM_ADDRESS not configured" 

137 ) 

138 return 

139 

140 port = int(os.environ.get("OPENHANGAR_SMTP_PORT", "587")) 

141 user = os.environ.get("OPENHANGAR_SMTP_USER", "").strip() 

142 password = _env_or_file("SMTP_PASSWORD") 

143 use_tls = os.environ.get("OPENHANGAR_SMTP_USE_TLS", "true").lower() not in ( 

144 "false", 

145 "0", 

146 "no", 

147 ) 

148 

149 msg = MIMEText(detail, "plain", "utf-8") 

150 msg["Subject"] = f"[OpenHangar] Security alert: {event_type}" 

151 msg["From"] = from_addr 

152 msg["To"] = to 

153 

154 conn = smtplib.SMTP(host, port, timeout=10) 

155 if use_tls: 

156 conn.ehlo() 

157 conn.starttls() 

158 conn.ehlo() 

159 if user: 

160 conn.login(user, password) 

161 conn.sendmail(from_addr, [to], msg.as_bytes()) 

162 conn.quit() 

163 except Exception as exc: # noqa: BLE001 -- one alert channel's failure must not affect others 

164 _log.error("Security alert: email delivery failed: %s", exc) 

165 

166 def _send_webhook(self, url: str, event_type: str, detail: str) -> None: 

167 try: 

168 payload = json.dumps({"event": event_type, "detail": detail}).encode( 

169 "utf-8" 

170 ) 

171 req = urllib.request.Request( 

172 url, 

173 data=payload, 

174 headers={"Content-Type": "application/json"}, 

175 method="POST", 

176 ) 

177 with urllib.request.urlopen(req, timeout=10): # nosec B310 # scheme restricted to http(s) at startup (OPENHANGAR_ALERT_WEBHOOK_URL check in init.py) 

178 pass 

179 except Exception as exc: # noqa: BLE001 -- one alert channel's failure must not affect others 

180 _log.error("Security alert: webhook delivery failed: %s", exc) 

181 

182 

183def attach_to_logger() -> None: 

184 """Attach the SecurityAlertHandler to the openhangar logger. Idempotent.""" 

185 logger = logging.getLogger("openhangar") 

186 if not any(isinstance(h, SecurityAlertHandler) for h in logger.handlers): 

187 logger.addHandler(SecurityAlertHandler())