From 7c0c43cc54c721b8a835ebaf6241cb735d4b9f7e Mon Sep 17 00:00:00 2001 From: "Leon Acosta @ Dandelion Labs" Date: Tue, 13 Oct 2026 12:16:16 +0200 Subject: [PATCH] Refresh internal KMS trust and deliver durable monitor alerts --- deploy/crypto-agility-alerts@.service | 25 +++ deploy/crypto-agility-alerts@.timer | 10 ++ docs/consumer-deployment.md | 40 ++++- tests/test_alert_delivery.py | 134 ++++++++++++++++ tests/test_kms_status.py | 123 +++++++++++++++ tools/alert_delivery.py | 219 ++++++++++++++++++++++++++ tools/consumer.py | 45 ++++-- tools/kms_status.py | 136 ++++++++++++++++ 8 files changed, 720 insertions(+), 12 deletions(-) create mode 100644 deploy/crypto-agility-alerts@.service create mode 100644 deploy/crypto-agility-alerts@.timer create mode 100644 tests/test_alert_delivery.py create mode 100644 tests/test_kms_status.py create mode 100644 tools/alert_delivery.py create mode 100644 tools/kms_status.py diff --git a/deploy/crypto-agility-alerts@.service b/deploy/crypto-agility-alerts@.service new file mode 100644 index 0000000..e6a2fd3 --- /dev/null +++ b/deploy/crypto-agility-alerts@.service @@ -0,0 +1,25 @@ +# Configure EnvironmentFile in a host-specific drop-in to read the existing mail +# runtime environment in place. Never place its values in this unit or a copy. +[Unit] +Description=Crypto-agility alert delivery (%i) +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +User=%i +Group=%i +WorkingDirectory=/opt/crypto-agility +ExecStart=/opt/crypto-agility/.venv/bin/python -m tools.alert_delivery --config /etc/crypto-agility/%i/delivery.json --state-directory /var/lib/crypto-agility-consumer/%i +TimeoutStartSec=120 +UMask=0077 +NoNewPrivileges=true +PrivateTmp=true +PrivateDevices=true +ProtectSystem=strict +ProtectHome=true +RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 +ReadWritePaths=/var/lib/crypto-agility-consumer/%i +ReadOnlyPaths=/etc/crypto-agility/%i +StandardOutput=journal +StandardError=journal diff --git a/deploy/crypto-agility-alerts@.timer b/deploy/crypto-agility-alerts@.timer new file mode 100644 index 0000000..e82825c --- /dev/null +++ b/deploy/crypto-agility-alerts@.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Retry pending crypto-agility alert delivery (%i) + +[Timer] +OnBootSec=30 +OnUnitInactiveSec=60 +Unit=crypto-agility-alerts@%i.service + +[Install] +WantedBy=timers.target diff --git a/docs/consumer-deployment.md b/docs/consumer-deployment.md index 4162e96..fd00a7f 100644 --- a/docs/consumer-deployment.md +++ b/docs/consumer-deployment.md @@ -54,8 +54,46 @@ The state directory contains `initialized.json`, `replay.json`, `state.json`, `c The long-running process retains the existing bounded in-memory authenticated-input cache and reevaluates expiration every cycle. After restart, a previously consumed token is rejected as replay; provision a new fresh observation and JTI. An unchanged token cannot silently become current assurance again. Historical trusted measurements and pending alerts survive, allowing later valid classical evidence to reveal a hybrid regression even after rejected inputs. -Reports go to stdout/journald; they are not alert-delivery receipts. The durable `state.json` outbox retains each occurrence's `deliveryId` until a separately configured delivery component confirms success. After confirmed delivery, acknowledge only that exact occurrence with `python -m tools.monitor --state /absolute/state/state.json --ack-alert `. Never acknowledge merely because a report was printed or read. No delivery adapter or destination is selected here. Input failures continue producing reports and durable pending alerts; failures that prevent persistence explicitly include `deliveryPersistenceFailure`. State-directory startup failures or stdout errors stop the CLI so the supervisor can restart it. The CLI does not exit on an assessment finding, avoiding replay-producing restart loops. +Reports go to stdout/journald; they are not alert-delivery receipts. The durable `state.json` outbox retains each occurrence's `deliveryId` until a separately configured delivery component confirms success. After confirmed delivery, acknowledge only that exact occurrence with `python -m tools.monitor --state /absolute/state/state.json --ack-alert `. Never acknowledge merely because a report was printed or read. The optional Resend adapter below handles the approved destination and provider receipt before acknowledgment. Input failures continue producing reports and durable pending alerts; failures that prevent persistence explicitly include `deliveryPersistenceFailure`. State-directory startup failures or stdout errors stop the CLI so the supervisor can restart it. The CLI does not exit on an assessment finding, avoiding replay-producing restart loops. `deploy/crypto-agility-consumer@.service` is an installation template, not discovered production coordinates. Review its paths and existing service account, install a pinned checkout and virtual environment at the chosen location, provision approved configuration/bundles, and retain private writable state. Install dependencies from `requirements.txt` with the supported Python version; ML-DSA verification additionally needs the compatible OpenSSL runtime described in the attestation documentation. The template grants no network access and exposes no signing credentials. Validate the unit on the actual Linux host with `systemd-analyze verify` before installing; no enable/start command is authorized by the mere presence of this template. Before activation, run the zero-skip acceptance suite, authenticate a fresh exact-bound bundle, exercise a selector switch, restart/replay rejection, expiry, missing configuration, durable failure recovery and confirmed alert delivery at the operator's destination. Verify actual service ownership, filesystem permissions and persistence across a service restart. Hosted CI must cover the final committed tree; local fixture tests alone do not establish production success. + +## Automated internal KMS status + +An operator can replace the manually asserted `keyStatus` object with the following internal policy. The existing `trust` entry must use the pinned ML-DSA-65 SPKI DER key; a matching key identifier alone is insufficient: + +```json +{ + "mode": "aws-kms", + "receipt": "/absolute/private/config/kms-status.json", + "withdrawal": "/absolute/private/config/withdrawal.json", + "keyArn": "arn:aws:kms:REGION:ACCOUNT:key/UUID", + "spkiSha256": "" +} +``` + +The independently provisioned withdrawal file contains `{"version":1,"withdrawn":false,"reason":""}`. Changing `withdrawn` to `true` immediately fails the next consumer check; a missing/malformed file fails closed. The refresher never writes this override. Retain the local override independently of downloaded receipts, bundles and deployment releases. + +Run `python -m tools.kms_status --policy /private/kms-policy.json --output /private/kms-status.json` in the existing authorized signing environment. The policy contains exactly `version: 1`, `keyArn`, `spkiSha256` and `ttlSeconds`. Use an explicit key ARN, never an alias. A 15-minute workflow interval with `ttlSeconds: 1800` leaves one refresh interval of margin; the maximum accepted receipt lifetime is one hour. Optional `--profile qk-attest` selects an already configured local profile. Actions should inherit its existing AWS authentication and omit the profile. No signing API is called and no credentials are read out, copied, exported to the consumer or installed on its host. + +The bounded CLI invokes authenticated `DescribeKey` and `GetPublicKey` against the fixed regional HTTPS KMS endpoint. It requires the exact ARN, enabled state, `ML_DSA_65`, `SIGN_VERIFY`, `ML_DSA_SHAKE_256`, valid ML-DSA-65 SPKI framing and the approved SHA-256 digest. AWS documents the SPKI and algorithm fields in [GetPublicKey](https://docs.aws.amazon.com/kms/latest/APIReference/API_GetPublicKey.html). Requests that fail or disagree replace the receipt with `unknown` and return a nonzero exit code. Transfer that unknown receipt too; do not conceal a disabled key by publishing only successful refreshes. Invalid local policy or an unwritable output path stops publication; an old receipt expires naturally and cannot remain valid indefinitely. + +Publish the receipt over the independently authenticated SSH connection into the private configuration directory, with atomic replacement, file/directory fsync and owner-only permissions. Its exact fields are `version`, `status`, `keyArn`, `spkiSha256`, `checkedAt`, `validUntil`, and `source` (`aws-kms:DescribeKey+GetPublicKey`). This JSON is not self-authenticating: trust depends on the authorized refresher, verified SSH host identity and protected local file. The consumer checks the receipt against both the configured ARN/pin and actual trusted public key, requires current bounded timestamps, and persists the highest valid KMS check time to reject rollback, including unknown-status receipts. After an unknown receipt, only an active check with a strictly later timestamp can restore acceptance. Receipt absence, expiry, unknown status or local withdrawal queues a durable monitor failure without replacing the last trusted baseline. + +This internal policy establishes that the authorized refresh observed the expected KMS key enabled, subject to the receipt's freshness window and AWS eventual consistency. It does not establish public, root-signed nonrevocation, third-party measurement independence or PQC TLS authentication. Public revocation publication and independent interoperability remain distinct acceptance work. + +## Durable email delivery + +The approved QuantaKrypto alert destination is `leon@quantakrypto.com`. Provision owner-private `delivery.json` with `{"version":1,"recipient":"leon@quantakrypto.com","sender":"quantakrypto "}` after confirming the existing sender configuration. Honor a configured production `EMAIL_FROM` when provisioning this sender instead of assuming the default. The adapter uses the existing `RESEND_API_KEY` from its runtime environment; it does not create, copy or print the key. Configure the service to read the existing host environment in place using a reviewed host-specific `EnvironmentFile` drop-in or an existing launcher that passes only the needed variable. Protect that existing source file with owner-only permissions before activation. The public contact form and its automatic confirmation are not a monitor transport. + +Explicitly initialize delivery once with `python -m tools.alert_delivery --initialize-delivery --state-directory /absolute/private/state`, after initializing the consumer state. This adds `delivery.json` and its stable lock inside the persistent state directory. Initialization refuses previous or interrupted delivery state. Missing delivery state after initialization is an error; never delete it to retry a send. Run `python -m tools.alert_delivery --config /absolute/private/config/delivery.json --state-directory /absolute/private/state` from the existing mail runtime environment. The service/timer templates run independently from the offline consumer, so mail failures do not stop measurement assessment. + +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. + +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. + +For production acceptance, send a clearly labeled monitor test occurrence to the approved recipient through this same durable path. Verify the provider-reported delivery event, persistent receipt and exact outbox acknowledgment, then exercise a send/status failure and recovery without losing its occurrence ID. Do not claim delivery from the CLI exit code alone. Hosted CI covers fixture provider behavior; the real host receipt establishes actual delivery. diff --git a/tests/test_alert_delivery.py b/tests/test_alert_delivery.py new file mode 100644 index 0000000..b2fdd04 --- /dev/null +++ b/tests/test_alert_delivery.py @@ -0,0 +1,134 @@ +"""Durable provider receipts, bounded idempotency and precise outbox acknowledgment.""" +import datetime as dt +import json +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch +from tools.consumer import initialize_state +from tools.state import enqueue_failure +from test_consumer import NOW, put + + +class Provider: + def __init__(self): + self.sent = [] + self.event = 'sent' + self.payload = None + self.failure = False + def request(self, method, path, payload=None, idempotency_key=None): + if method == 'POST': + self.sent.append(idempotency_key) + self.payload = payload + if self.failure: raise RuntimeError('provider timeout with sensitive detail') + return {'id': 'receipt-1'} + return {'id': 'receipt-1', 'to': self.payload['to'], 'from': self.payload['from'], 'subject': self.payload['subject'], 'last_event': self.event} + + +class DeliveryTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name).resolve() + self.state = self.root/'state' + self.state.mkdir(mode=0o700) + initialize_state(self.state, NOW) + 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.provider = Provider() + from tools.alert_delivery import initialize_delivery + initialize_delivery(self.state, NOW) + + def deliver(self, now=NOW): + from tools.alert_delivery import deliver_pending + return deliver_pending(self.state, self.config, self.provider, now=now) + + def pending_alerts(self): + return json.loads((self.state/'state.json').read_bytes())['pendingAlerts'] + + def test_acceptance_is_not_delivery_and_restart_polls_saved_receipt(self): + self.deliver() + self.assertTrue(self.pending_alerts()) + self.assertEqual(len(self.provider.sent), 1) + self.provider.event = 'delivered' + self.deliver() + self.assertEqual(len(self.provider.sent), 1) + self.assertEqual(self.pending_alerts(), []) + self.assertIn('restore approved hybrid', self.provider.payload['text']) + self.assertEqual(self.provider.payload['to'], ['leon@quantakrypto.com']) + + def test_failed_send_retains_same_occurrence_and_idempotency_key(self): + self.provider.failure = True + self.deliver() + self.deliver(NOW+dt.timedelta(seconds=60)) + self.assertEqual(self.provider.sent[0], self.provider.sent[1]) + self.assertEqual(self.pending_alerts()[0]['deliveryId'], self.pending['deliveryId']) + self.assertNotIn('sensitive detail', (self.state/'delivery.json').read_text()) + + def test_ambiguous_send_beyond_idempotency_window_is_not_blindly_resent(self): + self.provider.failure = True + self.deliver() + results = self.deliver(NOW+dt.timedelta(hours=24)) + self.assertEqual(len(self.provider.sent), 1) + self.assertEqual(results[0]['status'], 'reconciliation-required') + self.assertTrue(self.pending_alerts()) + + def test_receipt_persisted_before_ack_and_old_receipt_cannot_ack_recurrence(self): + self.provider.event = 'delivered' + with patch('tools.alert_delivery.acknowledge_alerts', side_effect=OSError('disk failure')): + self.deliver() + self.assertTrue(self.pending_alerts()) + self.deliver() + self.assertEqual(self.pending_alerts(), []) + self.assertEqual(len(self.provider.sent), 1) + new = enqueue_failure(self.state/'state.json', self.alert)[0] + self.assertNotEqual(new['deliveryId'], self.pending['deliveryId']) + self.provider.event = 'sent' + self.deliver() + self.assertEqual(self.pending_alerts()[0]['deliveryId'], new['deliveryId']) + self.assertEqual(len(self.provider.sent), 2) + + def test_bounce_and_mismatched_provider_receipt_never_acknowledge(self): + self.provider.event = 'bounced' + result = self.deliver() + self.assertEqual(result[0]['status'], 'delivery-failed') + self.assertTrue(self.pending_alerts()) + self.provider.event = 'delivered' + with patch.object(self.provider, 'request', return_value={'id': 'wrong', 'last_event': 'delivered'}): + self.deliver() + self.assertTrue(self.pending_alerts()) + + def test_missing_receipt_journal_or_clock_rollback_never_resets_delivery(self): + self.deliver() + with self.assertRaises(ValueError): self.deliver(NOW-dt.timedelta(seconds=1)) + (self.state/'delivery.json').unlink() + with self.assertRaises(ValueError): self.deliver() + self.assertFalse((self.state/'delivery.json').exists()) + self.assertTrue(self.pending_alerts()) + + def test_no_network_without_api_key(self): + from tools.alert_delivery import ResendClient + with self.assertRaises(ValueError): ResendClient('') + + 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))) + self.provider.failure = True + self.deliver() + first = set(self.provider.sent) + self.deliver(NOW+dt.timedelta(seconds=60)) + self.assertEqual(len(set(self.provider.sent)), 4) + self.assertEqual(len(first), 3) + + def test_corrupted_delivery_clock_is_not_accepted(self): + path = self.state/'delivery.json' + raw = json.loads(path.read_bytes()) + raw['lastNow'] = float('nan') + put(path, json.dumps(raw).encode()) + with self.assertRaises(ValueError): self.deliver() + self.assertEqual(self.provider.sent, []) + + +if __name__ == '__main__': unittest.main() diff --git a/tests/test_kms_status.py b/tests/test_kms_status.py new file mode 100644 index 0000000..2aeb5aa --- /dev/null +++ b/tests/test_kms_status.py @@ -0,0 +1,123 @@ +"""Authenticated KMS status policy and pinned consumer integration.""" +import base64 +import copy +import datetime as dt +import hashlib +import json +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch +from tools.mldsa import SPKI_PREFIX +import test_consumer as fixtures +NOW, put = fixtures.NOW, fixtures.put + +ARN = 'arn:aws:kms:eu-central-1:123456789012:key/11111111-1111-4111-8111-111111111111' +DER = SPKI_PREFIX + bytes(1952) +PIN = hashlib.sha256(DER).hexdigest() +POLICY = {'version': 1, 'keyArn': ARN, 'spkiSha256': PIN, 'ttlSeconds': 1800} +DESCRIBE = {'KeyMetadata': {'Arn': ARN, 'KeyState': 'Enabled', 'Enabled': True, 'KeySpec': 'ML_DSA_65', 'KeyUsage': 'SIGN_VERIFY'}} +PUBLIC = {'KeyId': ARN, 'KeySpec': 'ML_DSA_65', 'KeyUsage': 'SIGN_VERIFY', 'SigningAlgorithms': ['ML_DSA_SHAKE_256'], 'PublicKey': base64.b64encode(DER).decode()} + + +class KmsStatusTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name).resolve() + self.policy = self.root/'policy.json' + self.output = self.root/'status.json' + put(self.policy, json.dumps(POLICY).encode()) + + def refresh(self, responses=None): + from tools.kms_status import refresh + responses = responses or [DESCRIBE, PUBLIC] + with patch('tools.kms_status.aws_request', side_effect=responses): + return refresh(self.policy, self.output, now=NOW) + + def test_pinned_active_status_is_short_lived_and_private(self): + receipt = self.refresh() + self.assertEqual(receipt['status'], 'active') + self.assertEqual(receipt['keyArn'], ARN) + self.assertEqual(receipt['spkiSha256'], PIN) + self.assertEqual(receipt['validUntil'], '2026-10-12T00:30:00Z') + self.assertEqual(self.output.stat().st_mode & 0o777, 0o600) + + def test_failure_overwrites_active_status_with_unknown(self): + self.refresh() + from tools.kms_status import refresh + with patch('tools.kms_status.aws_request', side_effect=RuntimeError('DO NOT LOG CREDENTIALS')): + receipt = refresh(self.policy, self.output, now=NOW) + self.assertEqual(receipt['status'], 'unknown') + self.assertNotIn('DO NOT LOG', self.output.read_text()) + + def test_disabled_alias_wrong_arn_algorithm_and_pin_fail_closed(self): + mutations = [('KeyState', 'Disabled'), ('Enabled', False), ('KeySpec', 'RSA_2048'), ('Arn', ARN+'other')] + for name, value in mutations: + with self.subTest(name=name): + meta = copy.deepcopy(DESCRIBE) + meta['KeyMetadata'][name] = value + self.assertEqual(self.refresh([meta, PUBLIC])['status'], 'unknown') + for field, value in [('KeyId', ARN+'other'), ('SigningAlgorithms', ['RSASSA_PSS_SHA_256']), ('PublicKey', base64.b64encode(DER[:-1]+b'x').decode())]: + with self.subTest(field=field): + public = dict(PUBLIC, **{field: value}) + self.assertEqual(self.refresh([DESCRIBE, public])['status'], 'unknown') + + def test_consumer_requires_fresh_receipt_matching_trust_and_local_override(self): + from tools.consumer import _config + fixture = fixtures.ConsumerTests() + fixture.setUp() + self.addCleanup(fixture.doCleanups) + receipt = self.refresh() + status_path, withdrawal = fixture.root/'kms.json', fixture.root/'withdrawal.json' + put(status_path, json.dumps(receipt).encode()) + put(withdrawal, b'{"version":1,"withdrawn":false,"reason":"operator allows internal monitoring"}') + fixture.config['trust']['issuers']['observatory']['keys']['probe'] = {'alg': 'ML-DSA-65', 'format': 'spki-der', 'publicKey': base64.urlsafe_b64encode(DER).rstrip(b'=').decode()} + fixture.config['keyStatus'] = {'mode': 'aws-kms', 'receipt': str(status_path), 'withdrawal': str(withdrawal), 'keyArn': ARN, 'spkiSha256': PIN} + fixture.save_config() + self.assertEqual(_config(fixture.config_path, NOW)['issuer'], 'observatory') + with self.assertRaises(ValueError): _config(fixture.config_path, NOW+dt.timedelta(minutes=30)) + put(withdrawal, b'{"version":1,"withdrawn":true,"reason":"operator withdrew this key"}') + with self.assertRaises(ValueError): _config(fixture.config_path, NOW) + withdrawal.unlink() + with self.assertRaises(OSError): _config(fixture.config_path, NOW) + put(withdrawal, b'{"version":1,"withdrawn":false,"reason":"operator allows internal monitoring"}') + newer = dict(receipt, status='unknown', checkedAt='2026-10-12T00:00:01Z') + put(status_path, json.dumps(newer).encode()) + fixture.consumer().check(NOW+dt.timedelta(seconds=1)) + put(status_path, json.dumps(receipt).encode()) + rollback = fixture.consumer().check(NOW+dt.timedelta(seconds=2)) + self.assertEqual(rollback['assessment']['status'], 'invalid') + self.assertIn('status rollback', rollback['assessment']['findings'][0]['message']) + put(status_path, json.dumps(dict(receipt, checkedAt=newer['checkedAt'])).encode()) + tied = fixture.consumer().check(NOW+dt.timedelta(seconds=3)) + self.assertEqual(tied['assessment']['status'], 'invalid') + put(status_path, json.dumps(dict(receipt, checkedAt='2026-10-12T00:00:02Z')).encode()) + later = fixture.consumer().check(NOW+dt.timedelta(seconds=4)) + self.assertNotEqual(later['assessment']['status'], 'invalid', later) + + def test_status_clock_rollback_and_unsafe_receipt_target_fail_closed(self): + self.refresh() + from tools.kms_status import refresh + with self.assertRaises(ValueError): refresh(self.policy, self.output, now=NOW-dt.timedelta(seconds=1)) + self.output.unlink() + self.output.symlink_to(self.policy) + with self.assertRaises((OSError, ValueError)): refresh(self.policy, self.output, now=NOW) + + def test_consumer_rejects_rollback_of_previously_accepted_status_receipt(self): + fixture = fixtures.ConsumerTests() + fixture.setUp() + self.addCleanup(fixture.doCleanups) + from tools.consumer import Consumer + config = dict(fixture.config, _kmsCheckedAt=NOW.timestamp()) + consumer = Consumer(fixture.config_path, fixture.bundles, fixture.state) + with patch('tools.consumer._config', return_value=config): + consumer.check(NOW) + config['_kmsCheckedAt'] -= 1 + with patch('tools.consumer._config', return_value=config): + result = consumer.check(NOW) + self.assertEqual(result['assessment']['status'], 'invalid') + self.assertIn('status rollback', result['assessment']['findings'][0]['message']) + + +if __name__ == '__main__': unittest.main() diff --git a/tools/alert_delivery.py b/tools/alert_delivery.py new file mode 100644 index 0000000..3a09f71 --- /dev/null +++ b/tools/alert_delivery.py @@ -0,0 +1,219 @@ +"""Durable Resend delivery of explicitly configured consumer alerts.""" +import argparse +import datetime as dt +from email.utils import parseaddr +import http.client +import json +import math +import multiprocessing +import os +from pathlib import Path +import re +import ssl + +from tools.consumer import _check_state, _directory +from tools.kms_status import private_json +from tools.state import acknowledge_alerts, canonical, locked_document +from tools.verifier import load_manifest + +DELIVERED = {'delivered', 'opened', 'clicked'} +FAILED = {'bounced', 'canceled', 'complained', 'failed', 'suppressed'} +IDEMPOTENCY_SECONDS = 23 * 3600 # Provider retains keys 24h; allow clock/network margin. + + +def _resend_worker(pipe, key, method, path, payload, idempotency_key): + connection = None + try: + connection = http.client.HTTPSConnection('api.resend.com', timeout=10, context=ssl.create_default_context()) + headers = {'Authorization': 'Bearer '+key, 'Content-Type': 'application/json', 'User-Agent': 'crypto-agility-monitor/1'} + if idempotency_key: + headers['Idempotency-Key'] = idempotency_key + connection.request(method, path, body=canonical(payload) if payload is not None else None, headers=headers) + response = connection.getresponse() + raw = response.read(65537) + if not 200 <= response.status < 300 or len(raw) > 65536: + raise ValueError('provider HTTP response rejected') + pipe.send(('ok', load_manifest(raw))) + except Exception: + pipe.send(('error', None)) + finally: + if connection: + connection.close() + pipe.close() + + +class ResendClient: + def __init__(self, api_key): + if not isinstance(api_key, str) or not 1 <= len(api_key) <= 2048 or any(ord(c) < 33 or ord(c) > 126 for c in api_key): + raise ValueError('existing RESEND_API_KEY is required') + self.key = api_key + + def request(self, method, path, payload=None, idempotency_key=None): + if not ((method == 'POST' and path == '/emails') or (method == 'GET' and re.fullmatch(r'/emails/[A-Za-z0-9-]{1,128}', path))): + raise ValueError('unsupported email operation') + context = multiprocessing.get_context('spawn') + receive, send = context.Pipe(duplex=False) + process = context.Process(target=_resend_worker, args=(send, self.key, method, path, payload, idempotency_key), daemon=True) + process.start() + send.close() + try: + if not receive.poll(15): + raise ValueError('provider request deadline exceeded') + outcome, result = receive.recv() + if outcome != 'ok': + raise ValueError('provider request failed') + return result + finally: + receive.close() + if process.is_alive(): process.terminate() + process.join(timeout=1) + if process.is_alive(): + process.kill() + process.join(timeout=1) + + +def initialize_delivery(directory, now=None): + directory = Path(directory) + _check_state(directory) + now = now or dt.datetime.now(dt.timezone.utc) + with _directory(directory) as fd: + if 'delivery.json.lock' in os.listdir(fd): + raise ValueError('delivery initialization refuses previous state or interrupted initialization') + with locked_document(directory/'delivery.json') as (previous, save): + if previous is not None: + raise ValueError('delivery state already exists') + save({'version': 1, 'lastNow': now.timestamp(), 'records': {}}) + + +def _configuration(path): + config = private_json(path) + if set(config) != {'version', 'recipient', 'sender'} or type(config['version']) is not int or config['version'] != 1: + raise ValueError('explicit delivery configuration required') + for key in ('recipient', 'sender'): + value = config[key] + if not isinstance(value, str) or not 1 <= len(value) <= 320 or any(ord(c) < 32 or ord(c) > 126 for c in value): + raise ValueError('invalid mail address') + _, address = parseaddr(value) + if not re.fullmatch(r'[A-Za-z0-9.!#$%&\x27*+/=?^_`{|}~-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}', address): + raise ValueError('invalid mail address') + if key == 'recipient' and value != address: + raise ValueError('one explicit recipient address required') + return config + + +def _payload(alert, state, config): + label = str(alert.get('type', 'monitor-failure')) + if not re.fullmatch('[a-z0-9-]{1,80}', label): + raise ValueError('invalid alert type') + 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: + 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 deliver_pending(directory, config_path, client, *, now=None): + directory = Path(directory) + now = now or dt.datetime.now(dt.timezone.utc) + _check_state(directory) + config = _configuration(config_path) + with locked_document(directory/'state.json', require_existing=True) as (state, _): + pending = state.get('pendingAlerts', []) + results = [] + # Serialize delivery separately from monitoring so network waits never block + # collection/assessment. The committed provider receipt precedes outbox ack. + with locked_document(directory/'delivery.json', require_existing=True) as (journal, save): + if (set(journal) != {'version', 'lastNow', 'records'} or journal['version'] != 1 + or type(journal['version']) is not int or type(journal['lastNow']) not in (int, float) + or not math.isfinite(journal['lastNow']) or not isinstance(journal['records'], dict) + or now.tzinfo is None or now.timestamp() < journal['lastNow']): + raise ValueError('invalid delivery state or clock rollback') + journal['lastNow'] = now.timestamp() + save(journal) + for record in journal['records'].values(): + if (not isinstance(record, dict) or 'firstAttempt' not in record + 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): + raise ValueError('invalid alert occurrence') + 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)} + journal['records'][identity] = record + save(journal) # Durable intent before any external request. + record['lastChecked'] = now.timestamp() + save(journal) + result = {'deliveryId': identity, 'status': 'pending'} + try: + payload = record['payload'] + if payload['to'] != [config['recipient']] or payload['from'] != config['sender']: + raise ValueError('delivery configuration changed; reconcile pending receipt') + if not record.get('providerId'): + if now.timestamp()-record['firstAttempt'] >= IDEMPOTENCY_SECONDS: + result['status'] = 'reconciliation-required' + results.append(result) + continue + response = client.request('POST', '/emails', payload, 'crypto-agility/'+identity) + provider_id = response.get('id') + if not isinstance(provider_id, str) or not re.fullmatch('[A-Za-z0-9-]{1,128}', provider_id): + raise ValueError('provider receipt missing') + record['providerId'] = provider_id + save(journal) + if not record.get('deliveredAt'): + response = client.request('GET', '/emails/'+record['providerId']) + if (response.get('id') != record['providerId'] or response.get('to') != payload['to'] + or response.get('from') != payload['from'] or response.get('subject') != payload['subject']): + raise ValueError('provider receipt does not match submitted message') + event = response.get('last_event') + if event not in DELIVERED | FAILED | {'queued', 'sent', 'scheduled', 'delivery_delayed'}: + raise ValueError('unknown provider delivery event') + record['lastEvent'] = event + if event in DELIVERED: + record['deliveredAt'] = now.timestamp() + save(journal) + if record.get('deliveredAt'): + acknowledge_alerts(directory/'state.json', [identity]) + result['status'] = 'delivered' + elif record.get('lastEvent') in FAILED: + result['status'] = 'delivery-failed' + except Exception: + # Do not serialize exception text, response bodies or secrets. + record['lastError'] = 'delivery-or-acknowledgment-failed' + save(journal) + result['status'] = 'retry-pending' + results.append(result) + return results + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--state-directory', type=Path, required=True) + parser.add_argument('--config', type=Path) + parser.add_argument('--initialize-delivery', action='store_true') + args = parser.parse_args() + os.umask(0o077) + if args.initialize_delivery: + if args.config: parser.error('initialize delivery separately from configuration') + initialize_delivery(args.state_directory) + return 0 + if not args.config: parser.error('explicit delivery configuration required') + try: + result = deliver_pending(args.state_directory, args.config, ResendClient(os.environ.get('RESEND_API_KEY', ''))) + print(json.dumps(result, separators=(',', ':'))) + return int(any(item['status'] not in ('pending', 'delivered') for item in result)) + except Exception: + # CLI diagnostics are deliberately fixed; inspect protected state locally. + print('{"status":"delivery-unavailable","pendingAlertsRetained":true}') + return 1 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/tools/consumer.py b/tools/consumer.py index 37668e3..fe5a26c 100644 --- a/tools/consumer.py +++ b/tools/consumer.py @@ -90,13 +90,6 @@ def _config(path, now): raise ValueError('invalid consumer identity') if _origin(config['subject']) != config['subject']: raise ValueError('canonical origin subject required') - status = config['keyStatus'] - if not isinstance(status, dict) or set(status) != {'status', 'checkedAt', 'validUntil', 'source'}: - raise ValueError('explicit operator key status required') - checked, until = parse_time(status['checkedAt']), parse_time(status['validUntil']) - if (status['status'] != 'active' or not checked <= now < until or (until-checked).total_seconds() > 86400 - or not isinstance(status['source'], str) or not 1 <= len(status['source'].strip()) <= 2048): - raise ValueError('operator key status absent, inactive, future, expired or unbounded') trust = config['trust'] if not isinstance(trust, dict) or set(trust) != {'issuers'} or set(trust['issuers']) != {config['issuer']}: raise ValueError('exact issuer trust required') @@ -106,6 +99,17 @@ def _config(path, now): or policy.get('purposes') != ['pqc-observation'] or policy.get('disabled', False) is not False or config['keyId'] in policy.get('revokedKeys', [])): raise ValueError('exact enabled key and consumer scopes required') + status = config['keyStatus'] + if isinstance(status, dict) and status.get('mode') == 'aws-kms': + from tools.kms_status import validate_status + config['_kmsCheckedAt'] = validate_status(status, policy['keys'][config['keyId']], now) + else: + if not isinstance(status, dict) or set(status) != {'status', 'checkedAt', 'validUntil', 'source'}: + raise ValueError('explicit operator key status required') + checked, until = parse_time(status['checkedAt']), parse_time(status['validUntil']) + if (status['status'] != 'active' or not checked <= now < until or (until-checked).total_seconds() > 86400 + or not isinstance(status['source'], str) or not 1 <= len(status['source'].strip()) <= 2048): + raise ValueError('operator key status absent, inactive, future, expired or unbounded') return config @@ -153,8 +157,8 @@ def _check_state(directory): # Missing or unsafe files cannot trigger implicit baseline/replay reset. for name in ('state.json', 'replay.json', 'clock.json'): value = load_manifest(_read(fd, name, 4 * 1024 * 1024)) - if name == 'clock.json' and (set(value) != {'lastNow'} or type(value['lastNow']) not in (int, float) - or not math.isfinite(value['lastNow'])): + if name == 'clock.json' and (not {'lastNow'} <= set(value) <= {'lastNow', 'lastKmsCheck', 'lastKmsUnavailable'} + or any(type(n) not in (int, float) or not math.isfinite(n) for n in value.values())): raise ValueError('invalid consumer clock state') @@ -164,6 +168,17 @@ def __init__(self, config, bundles, state_directory): self.schema = json.loads((ROOT/'schema/crypto-agility-v2.schema.json').read_bytes()) self.cache = {} + def _record_kms_check(self, checked_at, *, unavailable=False): + with locked_document(self.directory/'clock.json', require_existing=True) as (clock, save): + if checked_at < clock.get('lastKmsCheck', 0): + raise ValueError('KMS status rollback') + if not unavailable and checked_at <= clock.get('lastKmsUnavailable', -1): + raise ValueError('newer KMS check required after unavailable status') + if unavailable: + clock['lastKmsUnavailable'] = checked_at + clock['lastKmsCheck'] = checked_at + save(clock) + def check(self, now=None): now = now or dt.datetime.now(dt.timezone.utc) state_safe = False @@ -173,8 +188,16 @@ def check(self, now=None): with locked_document(self.directory/'clock.json', require_existing=True) as (clock, save): if now.tzinfo is None or (clock and now.timestamp() < clock['lastNow']): raise ValueError('consumer clock rollback') - save({'lastNow': now.timestamp()}) - config = _config(self.config_path, now) + clock['lastNow'] = now.timestamp() + save(clock) + from tools.kms_status import KmsStatusUnavailable + try: + config = _config(self.config_path, now) + except KmsStatusUnavailable as unavailable: + self._record_kms_check(unavailable.checked_at, unavailable=True) + raise + if '_kmsCheckedAt' in config: + self._record_kms_check(config['_kmsCheckedAt']) values = _bundle(self.bundles) manifest = load_manifest(values['manifest.json']) if manifest.get('subject', {}).get('kind') != 'origin' or manifest['subject'].get('id') != config['subject']: diff --git a/tools/kms_status.py b/tools/kms_status.py new file mode 100644 index 0000000..6c646b5 --- /dev/null +++ b/tools/kms_status.py @@ -0,0 +1,136 @@ +"""Refresh internal trust status through authenticated AWS KMS; no signing.""" +import argparse +import base64 +import datetime as dt +import hashlib +from pathlib import Path +import re + +from tools.attestation import _decode +from tools.mldsa import MLDSA65Algorithm +from tools.state import locked_document +from tools.tls_probe import _run_bounded +from tools.verifier import load_manifest, parse_time + +SOURCE = 'aws-kms:DescribeKey+GetPublicKey' +ARN = re.compile(r'arn:aws:kms:([a-z]{2}-[a-z]+-\d):\d{12}:key/[a-f0-9-]{36}') + + +class KmsStatusUnavailable(ValueError): + def __init__(self, checked_at): + super().__init__('authenticated KMS check did not establish active status') + self.checked_at = checked_at + + +def private_json(path): + from tools.consumer import _directory, _read + path = Path(path) + with _directory(path.parent) as parent: + return load_manifest(_read(parent, path.name, 1048576)) + + +def _identity(arn, pin): + if not isinstance(arn, str) or not ARN.fullmatch(arn) or not isinstance(pin, str) or not re.fullmatch('[a-f0-9]{64}', pin): + raise ValueError('explicit KMS key ARN and SHA-256 SPKI pin required') + + +def aws_request(operation, arn, profile=None): + region = ARN.fullmatch(arn)[1] + # A fixed TLS endpoint overrides configurable service endpoints. Credentials + # remain in their existing AWS environment/profile; never read/export them. + args = ['aws', '--no-cli-pager', '--output', 'json', '--region', region, + '--endpoint-url', 'https://kms.'+region+'.amazonaws.com', + '--cli-connect-timeout', '5', '--cli-read-timeout', '10'] + if profile: + if not re.fullmatch('[A-Za-z0-9_.-]{1,128}', profile): + raise ValueError('invalid AWS profile name') + args += ['--profile', profile] + if operation not in ('describe-key', 'get-public-key'): + raise ValueError('unsupported KMS status operation') + code, raw = _run_bounded(args+['kms', operation, '--key-id', arn], 20, 65536) + if code: + raise ValueError('authenticated KMS status request failed') + return load_manifest(raw) + + +def refresh(policy_path, output_path, *, profile=None, now=None): + from tools.consumer import _directory + now = now or dt.datetime.now(dt.timezone.utc) + policy = private_json(policy_path) + if set(policy) != {'version', 'keyArn', 'spkiSha256', 'ttlSeconds'} or type(policy['version']) is not int or policy['version'] != 1: + raise ValueError('explicit KMS status policy required') + _identity(policy['keyArn'], policy['spkiSha256']) + ttl = policy['ttlSeconds'] + if type(ttl) is not int or not 60 <= ttl <= 3600 or now.tzinfo is None: + raise ValueError('bounded status lifetime and aware clock required') + output_path = Path(output_path) + with _directory(output_path.parent): + pass + with locked_document(output_path) as (previous, save): + if previous and now < parse_time(previous['checkedAt']): + raise ValueError('KMS status clock rollback') + stamp = lambda value: value.astimezone(dt.timezone.utc).isoformat(timespec='seconds').replace('+00:00', 'Z') + receipt = {'version': 1, 'status': 'unknown', 'keyArn': policy['keyArn'], 'spkiSha256': policy['spkiSha256'], + 'checkedAt': stamp(now), 'validUntil': stamp(now+dt.timedelta(seconds=ttl)), 'source': SOURCE} + try: + meta = aws_request('describe-key', policy['keyArn'], profile)['KeyMetadata'] + public = aws_request('get-public-key', policy['keyArn'], profile) + if (meta['Arn'] != policy['keyArn'] or meta['KeyState'] != 'Enabled' or meta['Enabled'] is not True + or meta['KeySpec'] != 'ML_DSA_65' or meta['KeyUsage'] != 'SIGN_VERIFY' + or public['KeyId'] != policy['keyArn'] or public['KeySpec'] != 'ML_DSA_65' + or public['KeyUsage'] != 'SIGN_VERIFY' or public['SigningAlgorithms'] != ['ML_DSA_SHAKE_256']): + raise ValueError('KMS status or algorithm mismatch') + der = base64.b64decode(public['PublicKey'], validate=True) + MLDSA65Algorithm().prepare_key(der) + if hashlib.sha256(der).hexdigest() != policy['spkiSha256']: + raise ValueError('KMS public key pin mismatch') + receipt['status'] = 'active' + except Exception: + # API/process output can contain sensitive environment details. Persist + # only an unknown status, never the exception or captured stderr. + pass + save(receipt) + return receipt + + +def validate_status(spec, key_policy, now): + if set(spec) != {'mode', 'receipt', 'withdrawal', 'keyArn', 'spkiSha256'} or spec['mode'] != 'aws-kms': + raise ValueError('invalid KMS consumer policy') + _identity(spec['keyArn'], spec['spkiSha256']) + override = private_json(spec['withdrawal']) + if (set(override) != {'version', 'withdrawn', 'reason'} or type(override['version']) is not int or override['version'] != 1 + or override['withdrawn'] is not False or not isinstance(override['reason'], str) or not 1 <= len(override['reason'].strip()) <= 2048): + raise ValueError('local trust is withdrawn or unspecified') + if not isinstance(key_policy, dict) or set(key_policy) != {'alg', 'format', 'publicKey'} or key_policy['alg'] != 'ML-DSA-65' or key_policy['format'] != 'spki-der': + raise ValueError('pinned ML-DSA-65 trust required for KMS status') + der = _decode(key_policy['publicKey']) + MLDSA65Algorithm().prepare_key(der) + if hashlib.sha256(der).hexdigest() != spec['spkiSha256']: + raise ValueError('consumer public key pin mismatch') + receipt = private_json(spec['receipt']) + if (set(receipt) != {'version', 'status', 'keyArn', 'spkiSha256', 'checkedAt', 'validUntil', 'source'} + or type(receipt['version']) is not int or receipt['version'] != 1 + or receipt['status'] not in ('active', 'unknown') or receipt['source'] != SOURCE + or receipt['keyArn'] != spec['keyArn'] or receipt['spkiSha256'] != spec['spkiSha256']): + raise ValueError('KMS receipt identity or status mismatch') + checked, until = parse_time(receipt['checkedAt']), parse_time(receipt['validUntil']) + if not checked <= now < until or not 0 < (until-checked).total_seconds() <= 3600: + raise ValueError('KMS status receipt is stale, future or unbounded') + if receipt['status'] != 'active': + raise KmsStatusUnavailable(checked.timestamp()) + return checked.timestamp() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--policy', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--profile') + args = parser.parse_args() + receipt = refresh(args.policy, args.output, profile=args.profile) + print('KMS internal status: '+receipt['status']) + return 0 if receipt['status'] == 'active' else 1 + + +if __name__ == '__main__': + raise SystemExit(main())