diff --git a/docs/consumer-deployment.md b/docs/consumer-deployment.md index 87542bf..13335a1 100644 --- a/docs/consumer-deployment.md +++ b/docs/consumer-deployment.md @@ -92,6 +92,10 @@ Explicitly initialize delivery once with `python -m tools.alert_delivery --initi Each alert occurrence gets a durable sending intent and immutable payload before any network call. The Resend idempotency key is derived from its `deliveryId`; a returned provider message ID is persisted before polling. The message names the subject, occurrence, finding and repair action. Up to three least-recently-checked pending occurrences are processed per run, so bounced or ambiguous messages cannot starve later alerts. The provider transport uses fixed HTTPS endpoints, rejects redirects and bounds response bytes and total request time. Errors expose only fixed diagnostics, never API credentials or response bodies. +Critical alerts (severity `error`, which includes every `monitor-failure`, and any `tls-regression`) are sent one occurrence per message as soon as they are queued. Other alerts wait until the oldest has been pending for an hour, then go out together as one digest of up to 20 alerts. A digest has its own durable record listing its member `deliveryId`s and its own idempotency key; its members stay pending until that one message is confirmed delivered, then they are acknowledged together. An unfinished digest keeps exactly its original members, so alerts that arrive meanwhile wait for the next one. + +A failure that persists across checks queues one occurrence. While it lasts, the same failure is queued again at most once a day as a reminder, instead of after every delivered occurrence. The next successful check forgets active failures, so a later recurrence alerts at once. + API acceptance (`sent` or `queued`) is not delivery. The adapter retains the pending alert until the provider's matching message record reports `delivered`, `opened` or `clicked`; it durably records that result before acknowledging the exact occurrence. This is provider-reported delivery to the receiving mail system, not proof that a human read or acted on the alert. The provider ID and delivery event remain in the private journal for acceptance evidence. Bounces, suppression, failure and complaints remain pending and report `delivery-failed` for operator action. Unknown or inaccessible provider status cannot acknowledge an alert; the existing key must permit message retrieval as well as sending. Resend retains [idempotency keys for 24 hours](https://resend.com/changelog/idempotency-keys). An ambiguous submission can retry the same immutable request for at most 23 hours. After that it reports `reconciliation-required` and never blindly resends. Reconcile the occurrence tag/provider activity before any operator recovery; the tool deliberately cannot guess whether a message was sent when both acknowledgment and provider ID were lost. Delivered receipts can be retried locally after an outbox-write failure without sending another email. Retain receipt history across releases; reaching the bounded journal capacity requires reviewed archival, not silent deletion. diff --git a/tests/test_alert_delivery.py b/tests/test_alert_delivery.py index b2fdd04..85e9b67 100644 --- a/tests/test_alert_delivery.py +++ b/tests/test_alert_delivery.py @@ -36,7 +36,7 @@ def setUp(self): self.config = self.root/'delivery.json' put(self.config, b'{"version":1,"recipient":"leon@quantakrypto.com","sender":"quantakrypto "}') self.alert = {'id': 'test-alert', 'type': 'tls-regression', 'severity': 'error', 'action': 'Inspect the TLS deployment and restore approved hybrid key exchange.'} - self.pending = enqueue_failure(self.state/'state.json', self.alert)[0] + self.pending = enqueue_failure(self.state/'state.json', self.alert, now=NOW)[0] self.provider = Provider() from tools.alert_delivery import initialize_delivery initialize_delivery(self.state, NOW) @@ -83,10 +83,11 @@ def test_receipt_persisted_before_ack_and_old_receipt_cannot_ack_recurrence(self self.deliver() self.assertEqual(self.pending_alerts(), []) self.assertEqual(len(self.provider.sent), 1) - new = enqueue_failure(self.state/'state.json', self.alert)[0] + # A persisting failure is reminded daily; the old receipt cannot acknowledge it. + new = enqueue_failure(self.state/'state.json', self.alert, now=NOW+dt.timedelta(hours=25))[0] self.assertNotEqual(new['deliveryId'], self.pending['deliveryId']) self.provider.event = 'sent' - self.deliver() + self.deliver(NOW+dt.timedelta(hours=25)) self.assertEqual(self.pending_alerts()[0]['deliveryId'], new['deliveryId']) self.assertEqual(len(self.provider.sent), 2) @@ -114,7 +115,7 @@ def test_no_network_without_api_key(self): def test_failed_first_batch_does_not_starve_later_occurrences(self): for number in range(3): - enqueue_failure(self.state/'state.json', dict(self.alert, id='other-'+str(number))) + enqueue_failure(self.state/'state.json', dict(self.alert, id='other-'+str(number)), now=NOW) self.provider.failure = True self.deliver() first = set(self.provider.sent) @@ -130,5 +131,54 @@ def test_corrupted_delivery_clock_is_not_accepted(self): with self.assertRaises(ValueError): self.deliver() self.assertEqual(self.provider.sent, []) + def test_persistent_failure_is_one_occurrence_until_daily_reminder(self): + self.provider.event = 'delivered' + self.deliver() + self.assertEqual(self.pending_alerts(), []) + for minutes in (1, 2, 60, 23*60): + later = NOW+dt.timedelta(minutes=minutes) + self.assertEqual(enqueue_failure(self.state/'state.json', self.alert, now=later), []) + self.deliver(later) + self.assertEqual(len(self.provider.sent), 1) + other = enqueue_failure(self.state/'state.json', dict(self.alert, id='different-failure'), now=NOW+dt.timedelta(minutes=5)) + self.assertEqual(other[0]['id'], 'different-failure') + reminder = enqueue_failure(self.state/'state.json', self.alert, now=NOW+dt.timedelta(hours=24)) + self.assertIn(self.alert['id'], [a['id'] for a in reminder]) + + def routine(self, number, minutes=0): + alert = {'id': 'routine-'+str(number), 'type': 'new-finding', 'severity': 'warning', 'code': 'fallback-risk', 'ref': 'r'+str(number), + 'action': 'Review the referenced evidence and refresh or correct the manifest.'} + return enqueue_failure(self.state/'state.json', alert, now=NOW+dt.timedelta(minutes=minutes))[0] + + def test_non_critical_alerts_wait_and_go_out_as_one_digest(self): + self.provider.event = 'delivered' + self.deliver() + self.assertEqual(len(self.provider.sent), 1) + first, second = self.routine(1), self.routine(2, 30) + self.deliver(NOW+dt.timedelta(minutes=59)) + self.assertEqual(len(self.provider.sent), 1) + self.assertEqual(len(self.pending_alerts()), 2) + self.provider.event = 'sent' + results = self.deliver(NOW+dt.timedelta(minutes=61)) + self.assertEqual(len(self.provider.sent), 2) + self.assertEqual(sorted(results[0]['members']), sorted([first['deliveryId'], second['deliveryId']])) + self.assertIn('digest (2 alerts)', self.provider.payload['subject']) + self.assertIn(first['deliveryId'], self.provider.payload['text']) + # A third alert arriving meanwhile does not change the unfinished digest. + self.routine(3, 62) + self.provider.event = 'delivered' + self.deliver(NOW+dt.timedelta(minutes=63)) + self.assertEqual(len(self.provider.sent), 2) + self.assertEqual([a['id'] for a in self.pending_alerts()], ['routine-3']) + + def test_critical_alert_is_not_held_for_the_digest(self): + self.provider.event = 'delivered' + self.deliver() + self.routine(1) + critical = enqueue_failure(self.state/'state.json', dict(self.alert, id='downgrade', severity='warning'), now=NOW)[0] + results = self.deliver(NOW+dt.timedelta(minutes=1)) + self.assertEqual([r['deliveryId'] for r in results], [critical['deliveryId']]) + self.assertEqual([a['id'] for a in self.pending_alerts()], ['routine-1']) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_cross_runtime.py b/tests/test_cross_runtime.py index 16f2b10..1d101d9 100644 --- a/tests/test_cross_runtime.py +++ b/tests/test_cross_runtime.py @@ -266,8 +266,9 @@ def cycle(instant): self.assertEqual(expired_manifest['assessment']['status'], 'invalid') retained_baseline = json.loads(original) after_failure = json.loads(remote_state.read_bytes()) - self.assertEqual({k: v for k, v in after_failure.items() if k != 'pendingAlerts'}, - {k: v for k, v in retained_baseline.items() if k != 'pendingAlerts'}) + bookkeeping = ('pendingAlerts', 'activeFailures') + self.assertEqual({k: v for k, v in after_failure.items() if k not in bookkeeping}, + {k: v for k, v in retained_baseline.items() if k not in bookkeeping}) self.assertTrue(any(a['type'] == 'monitor-failure' for a in after_failure['pendingAlerts'])) self.assertEqual(expired_manifest['changes']['alerts'][0]['type'], 'monitor-failure') diff --git a/tools/alert_delivery.py b/tools/alert_delivery.py index 3a09f71..dab4435 100644 --- a/tools/alert_delivery.py +++ b/tools/alert_delivery.py @@ -1,6 +1,7 @@ """Durable Resend delivery of explicitly configured consumer alerts.""" import argparse import datetime as dt +import hashlib from email.utils import parseaddr import http.client import json @@ -19,6 +20,12 @@ DELIVERED = {'delivered', 'opened', 'clicked'} FAILED = {'bounced', 'canceled', 'complained', 'failed', 'suppressed'} IDEMPOTENCY_SECONDS = 23 * 3600 # Provider retains keys 24h; allow clock/network margin. +# Critical alerts are sent one by one as soon as they are queued. The rest wait until +# the oldest has been pending for DIGEST_SECONDS and then go out together in one digest. +CRITICAL_TYPES = {'tls-regression'} +DIGEST_SECONDS = 3600 +DIGEST_LIMIT = 20 +MESSAGE_LIMIT = 16384 def _resend_worker(pipe, key, method, path, payload, idempotency_key): @@ -101,6 +108,18 @@ def _configuration(path): return config +def _critical(alert): + return alert.get('severity') == 'error' or alert.get('type') in CRITICAL_TYPES + + +def _queued(alert): + # Alerts queued before digests existed carry no time and are treated as due. + try: + return dt.datetime.fromisoformat(alert['queuedAt']).timestamp() + except (KeyError, TypeError, ValueError): + return 0.0 + + def _payload(alert, state, config): label = str(alert.get('type', 'monitor-failure')) if not re.fullmatch('[a-z0-9-]{1,80}', label): @@ -108,12 +127,39 @@ def _payload(alert, state, config): text = ('Crypto-agility monitoring alert\n\nSubject: '+str(state.get('subject') or 'unavailable')+ '\nOccurrence: '+alert['deliveryId']+'\nAction: '+str(alert.get('action', 'Review the monitor findings.'))+ '\n\nDetails:\n'+json.dumps(alert, sort_keys=True, indent=2)+'\n') - if len(text.encode()) > 16384: + if len(text.encode()) > MESSAGE_LIMIT: raise ValueError('alert message size limit') return {'from': config['sender'], 'to': [config['recipient']], 'subject': '[QuantaKrypto] Crypto-agility '+label, 'text': text, 'tags': [{'name': 'monitor', 'value': 'crypto-agility'}, {'name': 'delivery', 'value': alert['deliveryId']}]} +def _digest_section(alert): + label = str(alert.get('type', 'monitor-failure')) + if not re.fullmatch('[a-z0-9-]{1,80}', label): + raise ValueError('invalid alert type') + return ('\n['+label+'] Occurrence: '+alert['deliveryId']+'\nAction: '+str(alert.get('action', 'Review the monitor findings.'))+ + '\nDetails:\n'+json.dumps(alert, sort_keys=True, indent=2)+'\n') + + +def _digest(alerts, state, config): + """Group the oldest non-critical alerts that fit one message; returns (id, members, payload).""" + head = 'Crypto-agility monitoring digest\n\nSubject: '+str(state.get('subject') or 'unavailable')+'\n' + members, body = [], '' + for alert in sorted(alerts, key=lambda a: (_queued(a), a['deliveryId'])): + section = _digest_section(alert) + if len(members) >= DIGEST_LIMIT or len((head+body+section).encode()) > MESSAGE_LIMIT: + break + members.append(alert['deliveryId']) + body += section + if not members: + raise ValueError('alert message size limit') + identity = hashlib.sha256(('digest:'+','.join(sorted(members))).encode()).hexdigest()[:32] + payload = {'from': config['sender'], 'to': [config['recipient']], + 'subject': '[QuantaKrypto] Crypto-agility digest ('+str(len(members))+' alerts)', + 'text': head+body, 'tags': [{'name': 'monitor', 'value': 'crypto-agility'}, {'name': 'delivery', 'value': identity}]} + return identity, members, payload + + def deliver_pending(directory, config_path, client, *, now=None): directory = Path(directory) now = now or dt.datetime.now(dt.timezone.utc) @@ -137,21 +183,38 @@ def deliver_pending(directory, config_path, client, *, now=None): or any(type(record[name]) not in (int, float) or not math.isfinite(record[name]) or record[name] > now.timestamp() for name in ('firstAttempt', 'lastChecked', 'deliveredAt') if name in record)): raise ValueError('invalid delivery receipt timestamp') - pending = sorted(pending, key=lambda alert: journal['records'].get(alert.get('deliveryId'), {}).get('lastChecked', -1))[:3] for alert in pending: - identity = alert.get('deliveryId', '') - if not isinstance(identity, str) or not re.fullmatch('[a-f0-9]{32}', identity): + if not isinstance(alert.get('deliveryId'), str) or not re.fullmatch('[a-f0-9]{32}', alert['deliveryId']): raise ValueError('invalid alert occurrence') + pending_ids = {alert['deliveryId'] for alert in pending} + urgent = sorted((a for a in pending if _critical(a)), + key=lambda alert: journal['records'].get(alert['deliveryId'], {}).get('lastChecked', -1))[:3] + work = [(alert['deliveryId'], [alert['deliveryId']], lambda alert=alert: _payload(alert, state, config)) for alert in urgent] + routine = [a for a in pending if not _critical(a)] + # An unfinished digest keeps its exact members until its receipt resolves. + open_digest = next((name for name, record in journal['records'].items() + if isinstance(record, dict) and record.get('members') and not record.get('deliveredAt') + and record.get('lastEvent') not in FAILED and set(record['members']) <= pending_ids), None) + if open_digest: + work.append((open_digest, journal['records'][open_digest]['members'], None)) + elif routine and now.timestamp() - min(_queued(a) for a in routine) >= DIGEST_SECONDS: + identity, members, payload = _digest(routine, state, config) + work.append((identity, members, lambda payload=payload: payload)) + for identity, members, build in work: record = journal['records'].get(identity) if record is None: if len(journal['records']) >= 10000: raise ValueError('delivery receipt capacity requires archival') - record = {'firstAttempt': now.timestamp(), 'payload': _payload(alert, state, config)} + record = {'firstAttempt': now.timestamp(), 'payload': build()} + if len(members) > 1 or members[0] != identity: + record['members'] = members journal['records'][identity] = record save(journal) # Durable intent before any external request. record['lastChecked'] = now.timestamp() save(journal) result = {'deliveryId': identity, 'status': 'pending'} + if record.get('members'): + result['members'] = record['members'] try: payload = record['payload'] if payload['to'] != [config['recipient']] or payload['from'] != config['sender']: @@ -180,7 +243,7 @@ def deliver_pending(directory, config_path, client, *, now=None): record['deliveredAt'] = now.timestamp() save(journal) if record.get('deliveredAt'): - acknowledge_alerts(directory/'state.json', [identity]) + acknowledge_alerts(directory/'state.json', record.get('members', [identity])) result['status'] = 'delivered' elif record.get('lastEvent') in FAILED: result['status'] = 'delivery-failed' diff --git a/tools/consumer.py b/tools/consumer.py index fe5a26c..89b990c 100644 --- a/tools/consumer.py +++ b/tools/consumer.py @@ -214,7 +214,7 @@ def check(self, now=None): try: if not state_safe: raise ValueError('consumer state directory is unsafe or unavailable') - changes['alerts'] = enqueue_failure(self.directory/'state.json', alert, require_existing=True) + changes['alerts'] = enqueue_failure(self.directory/'state.json', alert, require_existing=True, now=now) except (OSError, ValueError, TypeError, KeyError, RecursionError) as error: changes['deliveryPersistenceFailure'] = str(error) return {'subject': None, 'checkedAt': now.isoformat(), 'assessment': {'status': 'invalid', 'findings': [finding]}, 'pairs': [], 'changes': changes} diff --git a/tools/monitor.py b/tools/monitor.py index eeaed53..4972e2c 100644 --- a/tools/monitor.py +++ b/tools/monitor.py @@ -152,7 +152,7 @@ def read_input(name, path, limit=1048576): report["changes"] = {"alerts": [alert], "resolved": []} if state_path: try: - report["changes"]["alerts"] = enqueue_failure(state_path, alert, require_existing=require_existing_state) + report["changes"]["alerts"] = enqueue_failure(state_path, alert, require_existing=require_existing_state, now=now) except (OSError, ValueError, TypeError, KeyError, RecursionError) as exc: # State failure cannot promise durability. Keep an explicit output # signal for the supervisor rather than silently losing the failure. diff --git a/tools/state.py b/tools/state.py index 21a624d..10f27f6 100644 --- a/tools/state.py +++ b/tools/state.py @@ -173,7 +173,7 @@ def update_state(path, report, *, require_existing=False): pending = {a["id"]: a for a in (previous or {}).get("pendingAlerts", [])} for alert in delta["alerts"]: if alert["id"] not in pending: - pending[alert["id"]] = {**alert, "deliveryId": uuid.uuid4().hex} + pending[alert["id"]] = {**alert, "deliveryId": uuid.uuid4().hex, "queuedAt": report["checkedAt"]} # Save alerts with the baseline before output or external delivery. Retry the # same deliveryId until the supervisor explicitly acknowledges delivery. historical = (previous or {}).get("lastAuthenticatedMeasurements", (previous or {}).get("measurements", [])) @@ -205,14 +205,29 @@ def acknowledge_alerts(path, delivery_ids): return len(pending) - len(remaining) -def enqueue_failure(path, alert, *, require_existing=False): - """Durably queue an input failure without advancing a successful baseline.""" +FAILURE_REMINDER_SECONDS = 24 * 3600 + + +def enqueue_failure(path, alert, *, require_existing=False, now=None): + """Durably queue an input failure without advancing a successful baseline. + + A failure that persists across checks queues one occurrence, then one reminder per + day while it lasts, instead of a new occurrence each check once the last one was + delivered. A successful check replaces the state and so forgets active failures.""" + now = now or dt.datetime.now(dt.timezone.utc) with locked_document(path, require_existing=require_existing) as (previous, save): state = previous if previous is not None else {"noBaseline": True} pending = {a["id"]: a for a in state.get("pendingAlerts", [])} - if alert["id"] not in pending: - pending[alert["id"]] = {**alert, "deliveryId": uuid.uuid4().hex} + active = state.get("activeFailures", {}) + active = {k: v for k, v in active.items() if type(v) in (int, float)} if isinstance(active, dict) else {} + last = active.get(alert["id"]) + recent = last is not None and 0 <= now.timestamp() - last < FAILURE_REMINDER_SECONDS + if alert["id"] not in pending and not recent: + pending[alert["id"]] = {**alert, "deliveryId": uuid.uuid4().hex, "queuedAt": now.isoformat()} + active[alert["id"]] = now.timestamp() state["pendingAlerts"] = list(pending.values()) + state["activeFailures"] = active save(state) # Lead with the current failure, followed by any other undelivered alerts. - return [pending[alert["id"]]] + [a for k, a in pending.items() if k != alert["id"]] + current = [pending[alert["id"]]] if alert["id"] in pending else [] + return current + [a for k, a in pending.items() if k != alert["id"]]