modified on local at 2026-07-12 23:13:09
| 1 | 1 | #!/usr/bin/perl |
| 2 | 2 | #====================================================================== |
| 3 | 3 | # DriftSense -- alert dispatcher |
| 4 | 4 | # |
| 5 | 5 | # Cron entry point (NOT a CGI). Walks FILE_CHANGES + SCHEMA_CHANGE rows |
| 6 | 6 | # inserted since the last cursor tick, evaluates every ACTIVE rule |
| 7 | 7 | # against each new row, and POSTs matching payloads to the configured |
| 8 | 8 | # webhook (Slack / Discord / generic HTTP). Every attempt is logged |
| 9 | 9 | # to ALERT_DELIVERIES for the operator's audit trail. |
| 10 | 10 | # |
| 11 | 11 | # Config: /etc/drift_sense/drift_sense.conf (via MODS::Config). |
| 12 | 12 | # Log: /var/log/drift_sense/alert_dispatch.log (via cron redirect). |
| 13 | 13 | #====================================================================== |
| 14 | 14 | use strict; |
| 15 | 15 | use warnings; |
| 16 | 16 | use lib '/var/www/vhosts/3dshawn.com/site1'; |
| 17 | 17 | use POSIX (); |
| 18 | 18 | use Symbol (); |
| 19 | 19 | use JSON::PP; |
| 20 | 20 | use MODS::Config; |
| 21 | 21 | use MODS::DBConnect; |
| 22 | 22 | use MODS::Webhook; |
| 23 | 23 | |
| 24 | 24 | $| = 1; |
| 25 | 25 | my $ts = POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime); |
| 26 | 26 | |
| 27 | 27 | my $cfg = MODS::Config->new; |
| 28 | 28 | my $db = MODS::DBConnect->new; |
| 29 | 29 | my $dbh = $db->db_connect or die "[$ts] cannot connect to DriftSense storage DB\n"; |
| 30 | 30 | |
| 31 | 31 | my $public_url = $cfg->settings('public_url') || 'https://watchtower.3dshawn.com'; |
| 32 | ||
| 33 | # ---- Quiet hours --------------------------------------------------- | |
| 34 | # Which monitors are in a quiet window right now? A window that straddles | |
| 35 | # midnight (e.g. 22:00 -> 06:00) is active when now >= start OR now < end. | |
| 36 | my %QUIET_MID; | |
| 37 | { | |
| 38 | my $probe = $db->db_readwrite($dbh, q~ | |
| 39 | SELECT 1 AS ok FROM INFORMATION_SCHEMA.COLUMNS | |
| 40 | WHERE TABLE_SCHEMA=DATABASE() | |
| 41 | AND TABLE_NAME='FILE_MONITOR_SETTINGS' | |
| 42 | AND COLUMN_NAME='quiet_hours_start' | |
| 43 | ~, __FILE__, __LINE__); | |
| 44 | if ($probe && $probe->{ok}) { | |
| 45 | my @rows = $db->db_readwrite_multiple($dbh, q~ | |
| 46 | SELECT file_monitor_list_id AS mid, | |
| 47 | TIME_TO_SEC(quiet_hours_start) AS start_s, | |
| 48 | TIME_TO_SEC(quiet_hours_end) AS end_s | |
| 49 | FROM FILE_MONITOR_SETTINGS | |
| 50 | WHERE quiet_hours_start IS NOT NULL | |
| 51 | AND quiet_hours_end IS NOT NULL | |
| 52 | ~, __FILE__, __LINE__); | |
| 53 | my $now_row = $db->db_readwrite($dbh, | |
| 54 | "SELECT TIME_TO_SEC(CURTIME()) AS now_s", __FILE__, __LINE__); | |
| 55 | my $now = $now_row ? $now_row->{now_s} : 0; | |
| 56 | foreach my $r (@rows) { | |
| 57 | my $s = $r->{start_s}; my $e = $r->{end_s}; | |
| 58 | my $quiet = ($s <= $e) ? ($now >= $s && $now < $e) | |
| 59 | : ($now >= $s || $now < $e); | |
| 60 | $QUIET_MID{$r->{mid}} = 1 if $quiet; | |
| 61 | } | |
| 62 | if (%QUIET_MID) { | |
| 63 | print "[$ts] quiet-hours active for monitors: " . join(',', sort keys %QUIET_MID) . "\n"; | |
| 64 | } | |
| 65 | } | |
| 66 | } | |
| 67 | sub _monitor_is_quiet { my $mid = shift; return $QUIET_MID{$mid || 0} ? 1 : 0; } | |
| 32 | 68 | |
| 33 | 69 | # ---- Fetch active rules --------------------------------------------- |
| 34 | 70 | my @rules = $db->db_readwrite_multiple($dbh, q~ |
| 35 | 71 | SELECT alert_rule_id, rule_name, match_kind, match_path_glob, |
| 36 | 72 | match_status_list, exclude_ts_only, |
| 37 | 73 | delivery_kind, delivery_url, delivery_email, |
| 38 | 74 | rate_limit_per_min, coalesce_window_s, |
| 39 | 75 | frequency_threshold, frequency_window_min, |
| 40 | 76 | UNIX_TIMESTAMP(last_frequency_fire_at) AS last_freq_fire_epoch |
| 41 | 77 | FROM ALERT_RULES |
| 42 | 78 | WHERE is_active = 1 |
| 43 | 79 | ~, __FILE__, __LINE__); |
| 44 | 80 | |
| 45 | 81 | unless (@rules) { |
| 46 | 82 | print "[$ts] no active alert rules, exiting\n"; |
| 47 | 83 | $db->db_disconnect($dbh); |
| 48 | 84 | exit 0; |
| 49 | 85 | } |
| 50 | 86 | |
| 51 | 87 | # ---- Fetch cursor --------------------------------------------------- |
| 52 | 88 | my %cursor; |
| 53 | 89 | foreach my $r ($db->db_readwrite_multiple($dbh, q~ |
| 54 | 90 | SELECT source_kind, last_seen_id FROM ALERT_CURSOR |
| 55 | 91 | ~, __FILE__, __LINE__)) { |
| 56 | 92 | $cursor{$r->{source_kind}} = $r->{last_seen_id}; |
| 57 | 93 | } |
| 58 | 94 | $cursor{file} //= 0; |
| 59 | 95 | $cursor{schema} //= 0; |
| 60 | 96 | |
| 61 | 97 | # ---- Pull new FILE_CHANGES since cursor ------------------------------ |
| 62 | 98 | my $file_max = $cursor{file}; |
| 63 | 99 | my @file_new = $db->db_readwrite_multiple($dbh, qq~ |
| 64 | 100 | SELECT fc.file_changes_id AS id, fc.file_name, fc.status, fc.blob_sha, |
| 65 | 101 | fc.is_ts_only, fc.date_time, |
| 66 | 102 | UNIX_TIMESTAMP(fc.date_time) AS ts_epoch, |
| 103 | fc.file_monitor_list_id AS mid, | |
| 67 | 104 | s.server_name |
| 68 | 105 | FROM FILE_CHANGES fc |
| 69 | 106 | LEFT JOIN SERVERS s ON s.server_id = fc.server_id |
| 70 | 107 | WHERE fc.file_changes_id > $cursor{file} |
| 71 | 108 | ORDER BY fc.file_changes_id ASC |
| 72 | 109 | LIMIT 500 |
| 73 | 110 | ~, __FILE__, __LINE__); |
| 74 | 111 | foreach my $r (@file_new) { |
| 75 | 112 | $file_max = $r->{id} if $r->{id} > $file_max; |
| 113 | } | |
| 114 | ||
| 115 | # Quiet-hours filter: still advance cursor past silenced rows so we don't | |
| 116 | # fire them once the window ends, but drop them from the per-rule loop. | |
| 117 | if (%QUIET_MID) { | |
| 118 | my $before = scalar @file_new; | |
| 119 | @file_new = grep { !_monitor_is_quiet($_->{mid}) } @file_new; | |
| 120 | my $dropped = $before - scalar @file_new; | |
| 121 | print "[$ts] quiet-hours filtered $dropped file change(s) out of alert dispatch\n" if $dropped; | |
| 76 | 122 | } |
| 77 | 123 | |
| 78 | 124 | # ---- Pull new SCHEMA_CHANGE since cursor ----------------------------- |
| 79 | 125 | my $schema_max = $cursor{schema}; |
| 80 | 126 | my @schema_new = $db->db_readwrite_multiple($dbh, qq~ |
| 81 | 127 | SELECT sc.schema_change_id AS id, sc.database_name, sc.table_name, |
| 82 | 128 | sc.changes, sc.is_ts_only, sc.change_datetime, |
| 83 | 129 | UNIX_TIMESTAMP(sc.change_datetime) AS ts_epoch, |
| 84 | 130 | s.server_name |
| 85 | 131 | FROM SCHEMA_CHANGE sc |
| 86 | 132 | LEFT JOIN SERVERS s ON s.server_id = sc.server_id |
| 87 | 133 | WHERE sc.schema_change_id > $cursor{schema} |
| 88 | 134 | ORDER BY sc.schema_change_id ASC |
| 89 | 135 | LIMIT 500 |
| 90 | 136 | ~, __FILE__, __LINE__); |
| 91 | 137 | foreach my $r (@schema_new) { |
| 92 | 138 | $schema_max = $r->{id} if $r->{id} > $schema_max; |
| 93 | 139 | } |
| 94 | 140 | |
| 95 | 141 | my $total_new = scalar(@file_new) + scalar(@schema_new); |
| 96 | 142 | if ($total_new == 0) { |
| 97 | 143 | print "[$ts] no new changes since cursor (file=$cursor{file} schema=$cursor{schema})\n"; |
| 98 | 144 | $db->db_disconnect($dbh); |
| 99 | 145 | exit 0; |
| 100 | 146 | } |
| 101 | 147 | |
| 102 | 148 | # ---- Evaluate + deliver --------------------------------------------- |
| 103 | 149 | # HTTP transport is MODS::Webhook (shells out to /usr/bin/curl) so we |
| 104 | 150 | # don't need LWP::Protocol::https / IO::Socket::SSL as CPAN installs. |
| 105 | 151 | |
| 106 | 152 | my $delivered = 0; |
| 107 | 153 | my $skipped = 0; |
| 108 | 154 | my $failed = 0; |
| 109 | 155 | |
| 110 | 156 | foreach my $row (@file_new) { _process_row($row, 'file'); } |
| 111 | 157 | foreach my $row (@schema_new) { _process_row($row, 'schema'); } |
| 112 | 158 | |
| 113 | 159 | # ---- Frequency-mode rules ------------------------------------------ |
| 114 | 160 | # For any rule where frequency_threshold > 0, we ignore the per-row loop |
| 115 | 161 | # above and instead: query how many file changes matching the rule's |
| 116 | 162 | # path glob happened in the last window_min minutes. If it crosses the |
| 117 | 163 | # threshold and we haven't fired within the same window, fire once with |
| 118 | 164 | # the top offender. |
| 119 | 165 | foreach my $rule (@rules) { |
| 120 | 166 | next unless ($rule->{frequency_threshold} || 0) > 0; |
| 121 | 167 | my $win = int($rule->{frequency_window_min} || 60); |
| 122 | 168 | my $thr = int($rule->{frequency_threshold}); |
| 123 | 169 | |
| 124 | 170 | # Re-fire cooldown: don't fire more than once per window |
| 125 | 171 | my $last = int($rule->{last_freq_fire_epoch} || 0); |
| 126 | 172 | if ($last && (time - $last) < $win * 60) { |
| 127 | 173 | next; |
| 128 | 174 | } |
| 129 | 175 | |
| 130 | 176 | # Find files whose match_path_glob-hit change count > threshold |
| 131 | 177 | my $glob = $rule->{match_path_glob} // ''; |
| 132 | 178 | my $glob_sql = ''; |
| 133 | 179 | if (length $glob) { |
| 134 | 180 | # Approx: SQL LIKE. Convert glob wildcards to SQL % / _. |
| 135 | 181 | my $like = $glob; |
| 136 | 182 | $like =~ s/\*\*/%/g; |
| 137 | 183 | $like =~ s/\*/%/g; |
| 138 | 184 | $like =~ s/\?/_/g; |
| 139 | 185 | my $q_like = $dbh->quote($like); |
| 140 | 186 | $glob_sql = " AND fc.file_name LIKE $q_like"; |
| 141 | 187 | } |
| 142 | 188 | my @hot = $db->db_readwrite_multiple($dbh, qq~ |
| 143 | 189 | SELECT fc.file_name, |
| 144 | 190 | COUNT(*) AS ct, |
| 145 | 191 | MAX(fc.file_changes_id) AS latest_id, |
| 146 | 192 | s.server_name |
| 147 | 193 | FROM FILE_CHANGES fc |
| 148 | 194 | LEFT JOIN SERVERS s ON s.server_id = fc.server_id |
| 149 | 195 | WHERE fc.date_time >= DATE_SUB(NOW(), INTERVAL $win MINUTE) |
| 150 | 196 | $glob_sql |
| 151 | 197 | GROUP BY fc.file_name |
| 152 | 198 | HAVING ct >= $thr |
| 153 | 199 | ORDER BY ct DESC |
| 154 | 200 | LIMIT 5 |
| 155 | 201 | ~, __FILE__, __LINE__); |
| 156 | 202 | next unless @hot; |
| 157 | 203 | |
| 158 | 204 | # Fire once with the top offender's info |
| 159 | 205 | my $top = $hot[0]; |
| 160 | 206 | my $fake_row = { |
| 161 | 207 | id => $top->{latest_id}, |
| 162 | 208 | file_name => $top->{file_name}, |
| 163 | 209 | server_name => $top->{server_name} // 'local', |
| 164 | 210 | date_time => POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime), |
| 165 | 211 | ts_epoch => time, |
| 166 | 212 | status => 'frequency_alert', |
| 167 | 213 | }; |
| 168 | 214 | my $payload = _build_payload($rule, $fake_row, 'file'); |
| 169 | 215 | # Enrich summary with count + threshold |
| 170 | 216 | my $extra = "frequency alert: $top->{file_name} changed $top->{ct} times in the last $win min (threshold: $thr)"; |
| 171 | 217 | if (ref($payload->{generic_webhook}) eq 'HASH') { |
| 172 | 218 | $payload->{generic_webhook}{frequency_summary} = $extra; |
| 173 | 219 | $payload->{generic_webhook}{frequency_count} = $top->{ct}; |
| 174 | 220 | $payload->{generic_webhook}{frequency_window} = $win; |
| 175 | 221 | $payload->{generic_webhook}{summary} = $extra; |
| 176 | 222 | } |
| 177 | 223 | my ($status, $http_code, $body, $err) = _deliver($rule, $payload); |
| 178 | 224 | _log_delivery($rule, $fake_row, 'file', $status, $http_code, $body, $err); |
| 179 | 225 | if ($status eq 'sent') { |
| 180 | 226 | $delivered++; |
| 181 | 227 | $db->db_readwrite($dbh, qq~ |
| 182 | 228 | UPDATE ALERT_RULES |
| 183 | 229 | SET fire_count = fire_count + 1, |
| 184 | 230 | last_fired_at = NOW(), |
| 185 | 231 | last_frequency_fire_at = NOW() |
| 186 | 232 | WHERE alert_rule_id = $rule->{alert_rule_id} |
| 187 | 233 | ~, __FILE__, __LINE__); |
| 188 | 234 | } else { |
| 189 | 235 | $failed++; |
| 190 | 236 | } |
| 191 | 237 | } |
| 192 | 238 | |
| 193 | 239 | # ---- Advance cursor atomically -------------------------------------- |
| 194 | 240 | if ($file_max > $cursor{file}) { |
| 195 | 241 | $db->db_readwrite($dbh, |
| 196 | 242 | "UPDATE ALERT_CURSOR SET last_seen_id = $file_max WHERE source_kind = 'file'", |
| 197 | 243 | __FILE__, __LINE__); |
| 198 | 244 | } |
| 199 | 245 | if ($schema_max > $cursor{schema}) { |
| 200 | 246 | $db->db_readwrite($dbh, |
| 201 | 247 | "UPDATE ALERT_CURSOR SET last_seen_id = $schema_max WHERE source_kind = 'schema'", |
| 202 | 248 | __FILE__, __LINE__); |
| 203 | 249 | } |
| 204 | 250 | |
| 205 | 251 | # ---- Pin-drift alerts (Feature 5 wave 4) --------------------------- |
| 206 | 252 | # For each Named Release with alert_delivery_url set, check whether any |
| 207 | 253 | # pin's current-latest SHA differs from the pinned SHA. Log the drift |
| 208 | 254 | # to PIN_DRIFT_LOG (with notified=0) and, if not yet notified, POST an |
| 209 | 255 | # alert to the release's delivery URL, then set notified=1. |
| 210 | 256 | foreach my $rel ($db->db_readwrite_multiple($dbh, q~ |
| 211 | 257 | SELECT named_release_id, name, alert_delivery_kind, alert_delivery_url |
| 212 | 258 | FROM NAMED_RELEASES |
| 213 | 259 | WHERE alert_delivery_kind IS NOT NULL |
| 214 | 260 | AND alert_delivery_kind != 'none' |
| 215 | 261 | AND alert_delivery_url IS NOT NULL |
| 216 | 262 | AND alert_delivery_url != '' |
| 217 | 263 | ~, __FILE__, __LINE__)) { |
| 218 | 264 | |
| 219 | 265 | my $rid = $rel->{named_release_id}; |
| 220 | 266 | my @drifts; |
| 221 | 267 | foreach my $pin ($db->db_readwrite_multiple($dbh, qq~ |
| 222 | 268 | SELECT p.pin_id, p.file_name, p.blob_sha AS pinned_sha, |
| 223 | 269 | fc.server_id |
| 224 | 270 | FROM NAMED_RELEASE_PINS p |
| 225 | 271 | LEFT JOIN FILE_CHANGES fc ON fc.file_changes_id = p.file_changes_id |
| 226 | 272 | WHERE p.named_release_id = $rid |
| 227 | 273 | ~, __FILE__, __LINE__)) { |
| 228 | 274 | my $q_file = $dbh->quote($pin->{file_name}); |
| 229 | 275 | my $sid = int($pin->{server_id} || 0); |
| 230 | 276 | my $cur = $db->db_readwrite($dbh, qq~ |
| 231 | 277 | SELECT blob_sha, file_changes_id FROM FILE_CHANGES |
| 232 | 278 | WHERE server_id = $sid AND file_name = $q_file |
| 233 | 279 | ORDER BY file_changes_id DESC LIMIT 1 |
| 234 | 280 | ~, __FILE__, __LINE__); |
| 235 | 281 | my $cur_sha = $cur ? $cur->{blob_sha} : ''; |
| 236 | 282 | next unless $cur_sha && $cur_sha ne $pin->{pinned_sha}; |
| 237 | 283 | |
| 238 | 284 | # Have we already logged + notified this (pin, current_sha) tuple? |
| 239 | 285 | my $q_pinned = $dbh->quote($pin->{pinned_sha}); |
| 240 | 286 | my $q_current = $dbh->quote($cur_sha); |
| 241 | 287 | my $seen = $db->db_readwrite($dbh, qq~ |
| 242 | 288 | SELECT drift_id, notified FROM PIN_DRIFT_LOG |
| 243 | 289 | WHERE pin_id = $pin->{pin_id} |
| 244 | 290 | AND current_sha = $q_current |
| 245 | 291 | ORDER BY drift_id DESC LIMIT 1 |
| 246 | 292 | ~, __FILE__, __LINE__); |
| 247 | 293 | if ($seen && $seen->{notified}) { |
| 248 | 294 | next; # already notified for this exact drift state |
| 249 | 295 | } |
| 250 | 296 | |
| 251 | 297 | # Log the drift |
| 252 | 298 | my $q_fname = $dbh->quote($pin->{file_name}); |
| 253 | 299 | my $cid = int($cur->{file_changes_id} || 0); |
| 254 | 300 | $db->db_readwrite($dbh, qq~ |
| 255 | 301 | INSERT INTO PIN_DRIFT_LOG |
| 256 | 302 | (named_release_id, pin_id, file_name, pinned_sha, current_sha, file_changes_id, notified) |
| 257 | 303 | VALUES |
| 258 | 304 | ($rid, $pin->{pin_id}, $q_fname, $q_pinned, $q_current, $cid, 0) |
| 259 | 305 | ~, __FILE__, __LINE__); |
| 260 | 306 | push @drifts, { |
| 261 | 307 | file_name => $pin->{file_name}, |
| 262 | 308 | pinned_sha => $pin->{pinned_sha}, |
| 263 | 309 | current_sha => $cur_sha, |
| 264 | 310 | change_id => $cid, |
| 265 | 311 | }; |
| 266 | 312 | } |
| 267 | 313 | |
| 268 | 314 | next unless @drifts; |
| 269 | 315 | |
| 270 | 316 | # Build a summary payload |
| 271 | 317 | my $summary = sprintf('Named release "%s" -- %d pinned file(s) have drifted', $rel->{name}, scalar @drifts); |
| 272 | 318 | my $public_url = $cfg->settings('public_url') || 'https://watchtower.3dshawn.com'; |
| 273 | 319 | my $files_list = join("\n", map { |
| 274 | 320 | " * $_->{file_name} (pinned: " . substr($_->{pinned_sha}, 0, 12) . |
| 275 | 321 | " -> current: " . substr($_->{current_sha}, 0, 12) . ")" |
| 276 | 322 | } @drifts); |
| 277 | 323 | my $link = "$public_url/restore_release.cgi?release_id=$rid"; |
| 278 | 324 | |
| 279 | 325 | my $slack_body = { |
| 280 | 326 | text => "*DriftSense pin drift*: $summary", |
| 281 | 327 | attachments => [{ |
| 282 | 328 | color => '#f43f5e', |
| 283 | 329 | fields => [ |
| 284 | 330 | { title => 'Release', value => $rel->{name}, short => 1 }, |
| 285 | 331 | { title => 'Drifted files', value => scalar(@drifts), short => 1 }, |
| 286 | 332 | { title => 'Details', value => $files_list, short => 0 }, |
| 287 | 333 | ], |
| 288 | 334 | actions => [{ type => 'button', text => 'Restore all pinned', url => $link }], |
| 289 | 335 | }], |
| 290 | 336 | }; |
| 291 | 337 | my $discord_body = { |
| 292 | 338 | content => "**DriftSense pin drift**: $summary", |
| 293 | 339 | embeds => [{ |
| 294 | 340 | title => $rel->{name}, |
| 295 | 341 | url => $link, |
| 296 | 342 | color => 16005694, |
| 297 | 343 | description => $files_list, |
| 298 | 344 | }], |
| 299 | 345 | }; |
| 300 | 346 | my $generic_body = { |
| 301 | 347 | source => 'drift_sense', |
| 302 | 348 | alert_type => 'pin_drift', |
| 303 | 349 | release_id => $rid, |
| 304 | 350 | release_name => $rel->{name}, |
| 305 | 351 | drifted_count=> scalar @drifts, |
| 306 | 352 | summary => $summary, |
| 307 | 353 | files => \@drifts, |
| 308 | 354 | restore_url => $link, |
| 309 | 355 | }; |
| 310 | 356 | |
| 311 | 357 | my $body_ref = $rel->{alert_delivery_kind} eq 'slack' ? $slack_body |
| 312 | 358 | : $rel->{alert_delivery_kind} eq 'discord' ? $discord_body |
| 313 | 359 | : $generic_body; |
| 314 | 360 | |
| 315 | 361 | my ($code, $rbody, $err) = MODS::Webhook::post_json( |
| 316 | 362 | $rel->{alert_delivery_url}, $body_ref, |
| 317 | 363 | timeout => 8, |
| 318 | 364 | user_agent => 'DriftSense/1.0 (pin-drift)', |
| 319 | 365 | ); |
| 320 | 366 | |
| 321 | 367 | if (!$err && $code >= 200 && $code < 400) { |
| 322 | 368 | # Mark all just-logged rows as notified |
| 323 | 369 | foreach my $d (@drifts) { |
| 324 | 370 | my $q_current = $dbh->quote($d->{current_sha}); |
| 325 | 371 | $db->db_readwrite($dbh, qq~ |
| 326 | 372 | UPDATE PIN_DRIFT_LOG SET notified = 1 |
| 327 | 373 | WHERE named_release_id = $rid |
| 328 | 374 | AND file_name = @{[ $dbh->quote($d->{file_name}) ]} |
| 329 | 375 | AND current_sha = $q_current |
| 330 | 376 | AND notified = 0 |
| 331 | 377 | ~, __FILE__, __LINE__); |
| 332 | 378 | } |
| 333 | 379 | $delivered++; |
| 334 | 380 | print "[$ts] pin-drift alert delivered for release '$rel->{name}' (" . scalar(@drifts) . " drifted)\n"; |
| 335 | 381 | } else { |
| 336 | 382 | $failed++; |
| 337 | 383 | print "[$ts] pin-drift alert FAILED for release '$rel->{name}': " . ($err // "HTTP $code") . "\n"; |
| 338 | 384 | } |
| 339 | 385 | |
| 340 | 386 | # Bookkeep the check timestamp |
| 341 | 387 | $db->db_readwrite($dbh, qq~ |
| 342 | 388 | UPDATE NAMED_RELEASES SET last_pin_drift_check_at = NOW() |
| 343 | 389 | WHERE named_release_id = $rid |
| 344 | 390 | ~, __FILE__, __LINE__); |
| 345 | 391 | } |
| 346 | 392 | |
| 347 | 393 | $db->db_disconnect($dbh); |
| 348 | 394 | print "[$ts] processed +$total_new new changes: $delivered delivered, $skipped skipped, $failed failed\n"; |
| 349 | 395 | exit 0; |
| 350 | 396 | |
| 351 | 397 | #--------------------------------------------------------------------- |
| 352 | 398 | sub _process_row { |
| 353 | 399 | my ($row, $kind) = @_; |
| 354 | 400 | foreach my $rule (@rules) { |
| 355 | 401 | next unless _rule_matches($rule, $row, $kind); |
| 356 | 402 | |
| 357 | 403 | # Rate limit check |
| 358 | 404 | if ($rule->{rate_limit_per_min} > 0) { |
| 359 | 405 | my $rlq = $db->db_readwrite($dbh, qq~ |
| 360 | 406 | SELECT COUNT(*) AS n FROM ALERT_DELIVERIES |
| 361 | 407 | WHERE alert_rule_id = $rule->{alert_rule_id} |
| 362 | 408 | AND fired_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) |
| 363 | 409 | AND delivery_status IN ('sent','pending') |
| 364 | 410 | ~, __FILE__, __LINE__); |
| 365 | 411 | if ($rlq && $rlq->{n} >= $rule->{rate_limit_per_min}) { |
| 366 | 412 | _log_delivery($rule, $row, $kind, 'skipped', undef, undef, |
| 367 | 413 | "rate limit ($rule->{rate_limit_per_min}/min) reached"); |
| 368 | 414 | $skipped++; |
| 369 | 415 | next; |
| 370 | 416 | } |
| 371 | 417 | } |
| 372 | 418 | |
| 373 | 419 | # Build payload + POST |
| 374 | 420 | my $payload = _build_payload($rule, $row, $kind); |
| 375 | 421 | my ($status, $http_code, $body, $err) = _deliver($rule, $payload); |
| 376 | 422 | _log_delivery($rule, $row, $kind, $status, $http_code, $body, $err); |
| 377 | 423 | |
| 378 | 424 | if ($status eq 'sent') { $delivered++; } |
| 379 | 425 | elsif ($status eq 'skipped') { $skipped++; } |
| 380 | 426 | else { $failed++; } |
| 381 | 427 | |
| 382 | 428 | # Bump the rule's fire_count + last_fired_at |
| 383 | 429 | if ($status eq 'sent') { |
| 384 | 430 | $db->db_readwrite($dbh, qq~ |
| 385 | 431 | UPDATE ALERT_RULES |
| 386 | 432 | SET fire_count = fire_count + 1, |
| 387 | 433 | last_fired_at = NOW() |
| 388 | 434 | WHERE alert_rule_id = $rule->{alert_rule_id} |
| 389 | 435 | ~, __FILE__, __LINE__); |
| 390 | 436 | } |
| 391 | 437 | } |
| 392 | 438 | } |
| 393 | 439 | |
| 394 | 440 | #--------------------------------------------------------------------- |
| 395 | 441 | sub _rule_matches { |
| 396 | 442 | my ($rule, $row, $kind) = @_; |
| 397 | 443 | |
| 398 | 444 | # Kind filter |
| 399 | 445 | return 0 if $rule->{match_kind} ne 'either' && $rule->{match_kind} ne $kind; |
| 400 | 446 | |
| 401 | 447 | # ts-only exclusion (files only) |
| 402 | 448 | return 0 if $kind eq 'file' && $rule->{exclude_ts_only} && $row->{is_ts_only}; |
| 403 | 449 | |
| 404 | 450 | # Path glob |
| 405 | 451 | my $glob = $rule->{match_path_glob} // ''; |
| 406 | 452 | if (length $glob) { |
| 407 | 453 | my $target = $kind eq 'file' |
| 408 | 454 | ? ($row->{file_name} // '') |
| 409 | 455 | : (($row->{database_name} // '') . '.' . ($row->{table_name} // '')); |
| 410 | 456 | my $re = _glob_to_regex($glob); |
| 411 | 457 | return 0 unless $target =~ $re; |
| 412 | 458 | } |
| 413 | 459 | |
| 414 | 460 | # Status filter (files only) |
| 415 | 461 | if ($kind eq 'file' && $rule->{match_status_list}) { |
| 416 | 462 | my %allowed = map { s/^\s+|\s+$//g; $_ => 1 } |
| 417 | 463 | split /,/, $rule->{match_status_list}; |
| 418 | 464 | return 0 unless $allowed{ $row->{status} // '' }; |
| 419 | 465 | } |
| 420 | 466 | |
| 421 | 467 | return 1; |
| 422 | 468 | } |
| 423 | 469 | |
| 424 | 470 | sub _glob_to_regex { |
| 425 | 471 | my $g = shift; |
| 426 | 472 | my $re = quotemeta $g; |
| 427 | 473 | # Handle glob patterns: escaped meta chars back to regex meaning |
| 428 | 474 | $re =~ s{\\\*\\\*}{.*}g; # ** -> .* |
| 429 | 475 | $re =~ s{\\\*}{[^/]*}g; # * -> [^/]* |
| 430 | 476 | $re =~ s{\\\?}{.}g; # ? -> . |
| 431 | 477 | return qr/^$re$/; |
| 432 | 478 | } |
| 433 | 479 | |
| 434 | 480 | #--------------------------------------------------------------------- |
| 435 | 481 | sub _build_payload { |
| 436 | 482 | my ($rule, $row, $kind) = @_; |
| 437 | 483 | |
| 438 | 484 | my $where = $kind eq 'file' |
| 439 | 485 | ? ($row->{file_name} // '(unknown file)') |
| 440 | 486 | : (($row->{database_name} // '?') . '.' . ($row->{table_name} // '?')); |
| 441 | 487 | my $when_utc = $row->{date_time} || $row->{change_datetime} || ''; |
| 442 | 488 | my $server = $row->{server_name} || 'local'; |
| 443 | 489 | my $status_h = $kind eq 'file' ? ($row->{status} // '') : 'DDL change'; |
| 444 | 490 | my $link_id = $row->{id}; |
| 445 | 491 | my $link = "$public_url/diff.cgi?kind=$kind&id=$link_id"; |
| 446 | 492 | |
| 447 | 493 | my $summary = $kind eq 'file' |
| 448 | 494 | ? "$status_h: $where on $server" |
| 449 | 495 | : "$where changed on $server"; |
| 450 | 496 | |
| 451 | 497 | my $slack = { |
| 452 | 498 | text => "*DriftSense*: $summary", |
| 453 | 499 | attachments => [{ |
| 454 | 500 | color => $kind eq 'schema' ? '#f59e0b' : '#14b8a6', |
| 455 | 501 | fields => [ |
| 456 | 502 | { title => 'What', value => $where, short => 0 }, |
| 457 | 503 | { title => 'When', value => $when_utc, short => 1 }, |
| 458 | 504 | { title => 'Server', value => $server, short => 1 }, |
| 459 | 505 | ], |
| 460 | 506 | actions => [{ |
| 461 | 507 | type => 'button', text => 'View diff', url => $link, |
| 462 | 508 | }], |
| 463 | 509 | }], |
| 464 | 510 | }; |
| 465 | 511 | my $discord = { |
| 466 | 512 | content => "**DriftSense**: $summary", |
| 467 | 513 | embeds => [{ |
| 468 | 514 | title => $where, |
| 469 | 515 | url => $link, |
| 470 | 516 | color => $kind eq 'schema' ? 16101915 : 1349286, # amber, teal |
| 471 | 517 | fields => [ |
| 472 | 518 | { name => 'When', value => $when_utc || '?', inline => \1 }, |
| 473 | 519 | { name => 'Server', value => $server, inline => \1 }, |
| 474 | 520 | ], |
| 475 | 521 | }], |
| 476 | 522 | }; |
| 477 | 523 | my $generic = { |
| 478 | 524 | source => 'drift_sense', |
| 479 | 525 | rule_id => $rule->{alert_rule_id}, |
| 480 | 526 | rule_name => $rule->{rule_name}, |
| 481 | 527 | kind => $kind, |
| 482 | 528 | summary => $summary, |
| 483 | 529 | target => $where, |
| 484 | 530 | server => $server, |
| 485 | 531 | when_utc => $when_utc, |
| 486 | 532 | ts_epoch => $row->{ts_epoch}, |
| 487 | 533 | diff_url => $link, |
| 488 | 534 | status => $status_h, |
| 489 | 535 | }; |
| 490 | 536 | |
| 491 | 537 | return { |
| 492 | 538 | slack => $slack, |
| 493 | 539 | discord => $discord, |
| 494 | 540 | generic_webhook => $generic, |
| 495 | 541 | summary => $summary, |
| 496 | 542 | }; |
| 497 | 543 | } |
| 498 | 544 | |
| 499 | 545 | #--------------------------------------------------------------------- |
| 500 | 546 | sub _deliver { |
| 501 | 547 | my ($rule, $payload) = @_; |
| 502 | 548 | my $kind = $rule->{delivery_kind}; |
| 503 | 549 | my $url = $rule->{delivery_url}; |
| 504 | 550 | |
| 505 | 551 | if ($kind eq 'email') { |
| 506 | 552 | my $to = $rule->{delivery_email} || ''; |
| 507 | 553 | return ('failed', undef, undef, 'no delivery_email configured') unless length $to; |
| 508 | 554 | my $summary = $payload->{summary} // 'DriftSense alert'; |
| 509 | 555 | my $target = ($payload->{generic_webhook} && $payload->{generic_webhook}{target}) // '(unknown)'; |
| 510 | 556 | my $server = ($payload->{generic_webhook} && $payload->{generic_webhook}{server}) // 'local'; |
| 511 | 557 | my $link = ($payload->{generic_webhook} && $payload->{generic_webhook}{diff_url}) // ''; |
| 512 | 558 | |
| 513 | 559 | my $mail_body = "DriftSense captured a change matching alert rule: $rule->{rule_name}\n\n" |
| 514 | 560 | . "Target : $target\n" |
| 515 | 561 | . "Server : $server\n" |
| 516 | 562 | . "Summary: $summary\n" |
| 517 | 563 | . ($link ? "Diff : $link\n" : '') |
| 518 | 564 | . "\n(Automated notification from DriftSense.)\n"; |
| 519 | 565 | |
| 520 | 566 | my $from = $cfg->settings('alert_from_email') || 'drift_sense@localhost'; |
| 521 | 567 | my $subj = "[DriftSense] " . substr($summary, 0, 120); |
| 522 | 568 | |
| 523 | 569 | my $sendmail = '/usr/sbin/sendmail'; |
| 524 | 570 | $sendmail = '/usr/lib/sendmail' unless -x $sendmail; |
| 525 | 571 | return ('failed', undef, undef, "no sendmail binary") unless -x $sendmail; |
| 526 | 572 | |
| 527 | 573 | require IPC::Open3; |
| 528 | 574 | my ($wtr, $rdr, $err_fh); |
| 529 | 575 | $err_fh = Symbol::gensym(); |
| 530 | 576 | my $pid = eval { IPC::Open3::open3($wtr, $rdr, $err_fh, $sendmail, '-t', '-i') }; |
| 531 | 577 | return ('failed', undef, undef, "spawn: $@") if $@ || !$pid; |
| 532 | 578 | binmode $wtr; |
| 533 | 579 | print {$wtr} "From: $from\r\n"; |
| 534 | 580 | print {$wtr} "To: $to\r\n"; |
| 535 | 581 | print {$wtr} "Subject: $subj\r\n"; |
| 536 | 582 | print {$wtr} "Content-Type: text/plain; charset=utf-8\r\n"; |
| 537 | 583 | print {$wtr} "\r\n"; |
| 538 | 584 | print {$wtr} $mail_body; |
| 539 | 585 | close $wtr; |
| 540 | 586 | do { local $/; <$rdr> // ''; }; |
| 541 | 587 | do { local $/; <$err_fh> // ''; }; |
| 542 | 588 | close $rdr; close $err_fh; |
| 543 | 589 | waitpid $pid, 0; |
| 544 | 590 | my $rc = $? >> 8; |
| 545 | 591 | return $rc == 0 |
| 546 | 592 | ? ('sent', 0, "delivered to $to", undef) |
| 547 | 593 | : ('failed', $rc, undef, "sendmail exit $rc"); |
| 548 | 594 | } |
| 549 | 595 | unless (length($url // '')) { |
| 550 | 596 | return ('failed', undef, undef, "no delivery URL configured"); |
| 551 | 597 | } |
| 552 | 598 | |
| 553 | 599 | my $body_ref = $kind eq 'slack' ? $payload->{slack} |
| 554 | 600 | : $kind eq 'discord' ? $payload->{discord} |
| 555 | 601 | : $payload->{generic_webhook}; |
| 556 | 602 | |
| 557 | 603 | # MODS::Webhook handles JSON encoding + curl transport; returns |
| 558 | 604 | # (http_code, response_body, error_message). |
| 559 | 605 | my ($code, $rbody, $err) = MODS::Webhook::post_json( |
| 560 | 606 | $url, $body_ref, |
| 561 | 607 | timeout => 8, |
| 562 | 608 | user_agent => 'DriftSense/1.0 (alert-dispatch)', |
| 563 | 609 | ); |
| 564 | 610 | $rbody //= ''; |
| 565 | 611 | $rbody = substr($rbody, 0, 500) if length($rbody) > 500; |
| 566 | 612 | |
| 567 | 613 | if (!$err && $code >= 200 && $code < 400) { |
| 568 | 614 | return ('sent', $code, $rbody, undef); |
| 569 | 615 | } |
| 570 | 616 | return ('failed', $code, $rbody, $err || "HTTP $code"); |
| 571 | 617 | } |
| 572 | 618 | |
| 573 | 619 | #--------------------------------------------------------------------- |
| 574 | 620 | sub _log_delivery { |
| 575 | 621 | my ($rule, $row, $kind, $status, $http_code, $body, $err) = @_; |
| 576 | 622 | my $summary = substr(($rule->{rule_name} // '') . ' :: ' . ($row->{file_name} |
| 577 | 623 | || ($row->{database_name} . '.' . $row->{table_name}) |
| 578 | 624 | || '?'), 0, 490); |
| 579 | 625 | my $q_summary = $dbh->quote($summary); |
| 580 | 626 | my $q_status = $dbh->quote($status); |
| 581 | 627 | my $q_body = $dbh->quote($body // ''); |
| 582 | 628 | my $q_err = $dbh->quote($err // ''); |
| 583 | 629 | my $q_code = defined($http_code) ? int($http_code) : 'NULL'; |
| 584 | 630 | |
| 585 | 631 | $db->db_readwrite($dbh, qq~ |
| 586 | 632 | INSERT INTO ALERT_DELIVERIES |
| 587 | 633 | (alert_rule_id, source_kind, source_id, match_summary, |
| 588 | 634 | delivery_status, http_status, http_response, error_message) |
| 589 | 635 | VALUES |
| 590 | 636 | ($rule->{alert_rule_id}, '$kind', $row->{id}, $q_summary, |
| 591 | 637 | $q_status, $q_code, $q_body, $q_err) |
| 592 | 638 | ~, __FILE__, __LINE__); |
| 593 | 639 | } |