From cbc387ab9132a157c27c5a2dbb507c485e14a1cf Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 11 Oct 2026 05:12:08 +0100 Subject: [PATCH 1/2] Check ordinary saved-policy notifications and restart in CI Reuse the real-embed application to verify authenticated live startup, narrow destination admission, monitoring-only token scope, ordinary alert delivery, persisted sessions and saved policy across a real process restart. Keep source/binary identity, single-submission mutation rules and owned child/listener cleanup in a bounded receipt. This prevents component passes from hiding application-path regressions without another build or host capability. Contract-Neutral: CI-only validation of existing authenticated notification and persistence behaviour; no product, permission, installer or release contract delta. Change-source: pulse-maintainer --- .github/workflows/build-and-test.yml | 20 ++ scripts/check_notification_runtime.py | 371 +++++++++++++++++++++ scripts/tests/test_notification_runtime.py | 163 +++++++++ 3 files changed, 554 insertions(+) create mode 100644 scripts/check_notification_runtime.py create mode 100644 scripts/tests/test_notification_runtime.py diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml index 6171decf69..58ce67ae61 100644 --- a/.github/workflows/build-and-test.yml +++ b/.github/workflows/build-and-test.yml @@ -319,6 +319,26 @@ jobs: if-no-files-found: warn retention-days: 7 + # Reuse the exact real-embed binary, rather than another build or mock + # lifecycle callback. This owns only disposable loopback endpoints/data: + # saved policy must reach ordinary alerts and survive a real restart. + - name: Check ordinary notification policy and restart + timeout-minutes: 3 + run: >- + python3 scripts/check_notification_runtime.py + --binary "$RUNNER_TEMP/pulse-real-embed" + --expected-source-sha "${{ github.sha }}" + --receipt "$RUNNER_TEMP/pulse-notification-runtime.json" + + - name: Retain bounded notification runtime receipt + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: notification-runtime-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/pulse-notification-runtime.json + if-no-files-found: warn + retention-days: 7 + # An audit service failure or real advisory stays gating, but it must not # suppress formatting, tests, type-checking, or production-build proof. - name: Require frontend dependency audit diff --git a/scripts/check_notification_runtime.py b/scripts/check_notification_runtime.py new file mode 100644 index 0000000000..4c7fe197f3 --- /dev/null +++ b/scripts/check_notification_runtime.py @@ -0,0 +1,371 @@ +#!/usr/bin/env python3 +"""An owned offline source application and webhook; not installed acceptance. + +All data, credentials and network endpoints are generated in this guest. +Never prints credentials, response bodies or application logs. Mutations are +single submissions; uncertain POST/PUT outcomes stop without replay. +""" +import argparse +import datetime +import hashlib +import http.cookiejar +import http.server +import json +import os +from pathlib import Path +import re +import secrets +import signal +import socket +import subprocess +import tempfile +import threading +import time +import urllib.error +import urllib.request + +PUBLIC_URL = 'https://notification-runtime.example.test' +LIMIT = 2 * 1024 * 1024 +# prepareWebhookData's ordinary event; resolution is a different path. +ORDINARY_EVENT = 'alert' + + +class Failure(RuntimeError): + pass + + +def require(condition, message): + if not condition: + raise Failure(message) + + +def free_port(): + with socket.socket() as sock: + sock.bind(('127.0.0.1', 0)) + return sock.getsockname()[1] + + +class Sink: + def __init__(self): + self.rows = [] + self.lock = threading.Lock() + self.error = None + owner = self + + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def do_POST(self): + try: + size = int(self.headers.get('Content-Length', '-1')) + require(0 < size <= 65536, 'sink request exceeded declared bound') + raw = self.rfile.read(size) + require(len(raw) == size, 'sink short read') + obj = json.loads(raw) + require(isinstance(obj, dict), 'sink non-object') + with owner.lock: + owner.rows.append({'body': obj, 'sha256': hashlib.sha256(raw).hexdigest(), + 'bytes': size, 'path': self.path}) + self.send_response(200) + self.send_header('Content-Length', '2') + self.end_headers() + self.wfile.write(b'OK') + except Exception as exc: + owner.error = type(exc).__name__ + self.send_response(400) + self.send_header('Content-Length', '0') + self.end_headers() + + self.server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler) + self.server.daemon_threads = False + self.port = self.server.server_port + self.thread = threading.Thread(target=self.server.serve_forever, name='owned-webhook') + self.thread.start() + + def snapshot(self): + with self.lock: + return list(self.rows) + + def close(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + require(not self.thread.is_alive(), 'sink thread did not join') + with socket.socket() as sock: + sock.settimeout(1) + require(sock.connect_ex(('127.0.0.1', self.port)) != 0, 'sink listener stayed open') + + +def delivery_checks(row, expected_name): + """Keep each asserted closed condition, not private/untrusted payload text.""" + body = row['body'] + return {'owned_path': row['path'] == '/owned-alert', + 'resource_identity': body.get('resourceName') == expected_name, + 'saved_public_url': body.get('instance') == PUBLIC_URL, + 'ordinary_event': body.get('event') == ORDINARY_EVENT, + 'cpu_metric': body.get('type') == 'cpu'} + + +def source_binding(binary, expected): + require(bool(re.fullmatch(r'[0-9a-f]{40}', expected)), 'invalid expected source identity') + actual = subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip() + require(actual == expected, 'checkout differs from expected source') + require(subprocess.run(['git', 'diff', '--quiet', 'HEAD'], check=False).returncode == 0, + 'checkout has tracked source changes') + version = subprocess.check_output(['go', 'version', '-m', str(binary)], text=True) + fields = {} + for line in version.splitlines(): + match = re.match(r'\s*build\s+(vcs(?:\.[a-z]+)?)=(.*)$', line) + if match: + fields[match[1]] = match[2] + require(fields.get('vcs') == 'git' and fields.get('vcs.revision') == expected and + fields.get('vcs.modified') == 'false', 'binary VCS identity is not clean exact source') + return {'checkout_sha': actual, 'binary_vcs_revision': fields['vcs.revision'], + 'binary_vcs_modified': False} + + +class Journey: + def __init__(self, binary, data, log, sink): + self.binary, self.data, self.log, self.sink = binary, data, log, sink + self.port = free_port() + self.origin = f'http://127.0.0.1:{self.port}' + self.cookies = http.cookiejar.CookieJar() + self.client = urllib.request.build_opener(urllib.request.ProxyHandler({}), + urllib.request.HTTPCookieProcessor(self.cookies)) + self.password = secrets.token_hex(24) + self.token = None + self.token_id = None + self.process = None + self.phases, self.exits, self.requests, self.deliveries = [], [], [], [] + self.delivery_observations = [] + self.active_request = None + + def phase(self, name): + self.phases.append(name) + print('PHASE=' + name, flush=True) + + def request(self, path, payload=None, method=None, expected=200, auth='session'): + headers = {'Content-Type': 'application/json', 'User-Agent': 'pulse-notification-policy-journey'} + client = self.client + if auth != 'session': + client = urllib.request.build_opener(urllib.request.ProxyHandler({})) + if auth == 'agent': + headers['Authorization'] = 'Bearer ' + self.token + if payload is not None and auth == 'session': + for cookie in self.cookies: + if cookie.name == 'pulse_csrf': + headers['X-CSRF-Token'] = cookie.value + req = urllib.request.Request(self.origin + path, headers=headers, method=method, + data=None if payload is None else json.dumps(payload).encode()) + self.active_request = {'path': path, 'method': req.get_method(), 'authentication': auth} + print('REQUEST_BEGIN=' + json.dumps(self.active_request, sort_keys=True), flush=True) + start = time.monotonic() + try: + response = client.open(req, timeout=30) + except urllib.error.HTTPError as exc: + response = exc + with response: + raw = response.read(LIMIT + 1) + obs = dict(self.active_request, status=response.status, + elapsed_seconds=round(time.monotonic() - start, 3)) + self.requests.append(obs) + self.active_request = None + print('REQUEST_END=' + json.dumps(obs, sort_keys=True), flush=True) + require(len(raw) <= LIMIT, 'HTTP response exceeded byte bound') + require(response.status == expected, f'{path} expected HTTP {expected}, observed {response.status}') + if expected >= 400: + return None + try: + return json.loads(raw) if raw else None + except json.JSONDecodeError: + raise Failure(path + ' successful response was not JSON') from None + + def start(self): + env = {'PATH': os.environ['PATH'], 'HOME': str(self.data), 'LANG': 'C.UTF-8', + 'PULSE_DATA_DIR': str(self.data), 'BIND_ADDRESS': '127.0.0.1', + 'FRONTEND_PORT': str(self.port), 'PULSE_METRICS_PORT': str(free_port()), + 'PULSE_AUTH_USER': 'runtime-check', 'PULSE_AUTH_PASS': self.password, + 'PULSE_MOCK_MODE': 'false', 'DISCOVERY_ENABLED': 'false', 'PULSE_TELEMETRY': 'false'} + self.process = subprocess.Popen([str(self.binary)], cwd=self.data, env=env, + stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True) + deadline = time.monotonic() + 45 + while time.monotonic() < deadline: + require(self.process.poll() is None, 'owned server exited before readiness') + try: + self.request('/api/resources', expected=401, auth='none') + return + except (urllib.error.URLError, TimeoutError, ConnectionError): + time.sleep(.2) + raise Failure('owned server readiness deadline expired') + + def stop(self): + if self.process is None: + return + child, self.process = self.process, None + started = time.monotonic() + if child.poll() is None: + os.killpg(child.pid, signal.SIGTERM) + forced = False + try: + code = child.wait(timeout=40) + except subprocess.TimeoutExpired: + forced = True + os.killpg(child.pid, signal.SIGKILL) + code = child.wait(timeout=10) + self.exits.append({'exit': code, 'forced': forced, 'waited': True, + 'wait_seconds': round(time.monotonic() - started, 3)}) + require(code == 0 and not forced, 'owned server shutdown not clean') + with socket.socket() as sock: + sock.settimeout(1) + require(sock.connect_ex(('127.0.0.1', self.port)) != 0, 'owned listener stayed open') + + def denied_destination(self): + self.request('/api/notifications/webhooks', + {'name': 'unlisted-loopback', 'url': f'http://127.0.0.2:{self.sink.port}/blocked', + 'enabled': True, 'method': 'POST', 'service': 'generic'}, expected=400) + + def assert_settings(self): + settings = self.request('/api/system/settings') + require(settings.get('publicURL') == PUBLIC_URL and + settings.get('webhookAllowedPrivateCIDRs') == '127.0.0.1/32', 'saved system policy changed') + cfg = self.request('/api/alerts/config') + require(cfg.get('enabled') and cfg.get('activationState') == 'active' and + cfg.get('agentDefaults', {}).get('cpu', {}).get('trigger') == 80, + 'saved threshold/activation policy changed') + + def report(self, identity): + payload = {'agent': {'id': 'agent-' + identity, 'version': '0.0.0-dev', 'type': 'unified', + 'commandsEnabled': False, 'intervalSeconds': 60}, + 'host': {'id': 'machine-' + identity, 'machineId': 'machine-' + identity, + 'hostname': 'notification-' + identity + '.example.test', + 'platform': 'linux', 'cpuCount': 1}, + 'timestamp': datetime.datetime.now(datetime.timezone.utc).isoformat(), + 'sequenceId': 'v1:notification-' + identity + ':1', + 'metrics': {'cpuUsagePercent': 95, 'memory': {'totalBytes': 1073741824, + 'usedBytes': 429496729, 'freeBytes': 644245095, 'usage': 40}}} + self.request('/api/agents/agent/report', payload, auth='agent') + deadline = time.monotonic() + 30 + expected_name = payload['host']['hostname'] + while time.monotonic() < deadline: + require(self.process.poll() is None, 'owned server exited during ordinary delivery') + rows = self.sink.snapshot() + matches = [row for row in rows if row['body'].get('resourceName') == expected_name] + if matches: + require(len(matches) == 1, 'same first alert delivered more than once') + row = matches[0] + checks = delivery_checks(row, expected_name) + self.delivery_observations.append({'resource': identity, 'checks': checks, + 'bytes': row['bytes'], 'sha256': row['sha256']}) + require(all(checks.values()), 'ordinary delivery checks failed: ' + + ','.join(sorted(key for key, passed in checks.items() if not passed))) + self.deliveries.append({'resource': identity, 'event': ORDINARY_EVENT, 'metric': 'cpu', + 'saved_public_url': True, 'count': len(matches), + 'bytes': row['bytes'], 'sha256': row['sha256']}) + self.phase('ordinary-first-alert-' + identity) + return + require(self.sink.error is None, 'owned destination rejected payload') + time.sleep(.2) + active = self.request('/api/alerts/active') + self.phase('ordinary-delivery-deadline-active-' + ('present' if bool(active) else 'empty')) + raise Failure('ordinary first-alert destination deadline expired') + + def run(self): + try: + self.start() + self.request('/api/login', {'username': 'runtime-check', 'password': self.password}) + require(any(c.name == 'pulse_csrf' for c in self.cookies), 'login missing CSRF cookie') + require(self.request('/api/resources?limit=500').get('data') == [], 'initial live estate not empty') + self.phase('authenticated-empty-live-startup') + self.request('/api/system/settings/update', {'publicURL': PUBLIC_URL, + 'webhookAllowedPrivateCIDRs': '127.0.0.1/32', 'telemetryEnabled': False}) + self.denied_destination() + self.request('/api/notifications/webhooks', {'name': 'owned-first-alert', + 'url': f'http://127.0.0.1:{self.sink.port}/owned-alert', 'enabled': True, + 'method': 'POST', 'service': 'generic', + 'template': '{"instance":"{{.Instance | jsonString}}","resourceName":"{{.ResourceName | jsonString}}","event":"{{.Event | jsonString}}","type":"{{.Type | jsonString}}"}'}) + cfg = self.request('/api/alerts/config') + cfg.update(enabled=True, activationState='active', flappingEnabled=False, + timeThresholds={'agent': 0}, metricTimeThresholds={'agent': {'cpu': 0}}, + metricEvaluationWindows={'agent': {'cpu': 0}}, + agentDefaults={'cpu': {'trigger': 80, 'clear': 70}}, + disableAllAgents=False, disableAllAgentsOffline=True) + cfg['schedule'].update(cooldown=0, initialNotify='webhook', notifyOnResolve=False, + grouping={'enabled': False, 'window': 0, 'byNode': False, 'byGuest': False}) + self.request('/api/alerts/config', cfg, method='PUT') + response = self.request('/api/security/tokens', {'name': 'owned-monitoring-only', + 'scopes': ['agent:report'], 'expiresIn': '1h'}) + self.token, self.token_id = response['token'], response['record']['id'] + require(response['record']['scopes'] == ['agent:report'], 'monitoring-only scope widened') + self.request('/api/system/settings', expected=403, auth='agent') + self.assert_settings() + require(self.sink.snapshot() == [], 'destination called before ordinary synthetic alert') + self.phase('saved-policy-and-unlisted-destination-refusal') + self.report('a') + self.stop() + self.phase('first-child-exit-and-listener-closure') + self.start() + # Original authenticated cookies and issued token: no relogin or token replacement. + self.assert_settings() + tokens = self.request('/api/security/tokens') + records = tokens.get('tokens', []) if isinstance(tokens, dict) else tokens + matches = [r for r in records if r.get('id') == self.token_id] + require(len(matches) == 1 and matches[0].get('scopes') == ['agent:report'], + 'issued monitoring-only token not retained at restart') + self.denied_destination() + self.phase('restart-session-token-and-policy-readback') + self.report('b') + self.stop() + self.phase('second-child-exit-and-listener-closure') + finally: + self.stop() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--binary', type=Path, required=True) + parser.add_argument('--expected-source-sha', required=True) + parser.add_argument('--receipt', type=Path) + args = parser.parse_args() + binary = args.binary.resolve(strict=True) + binding = source_binding(binary, args.expected_source_sha) + report = {'source_binding': binding, 'kind': 'offline-development-source-notification-runtime', + 'binary_sha256': hashlib.sha256(binary.read_bytes()).hexdigest(), 'outcome': 'failed'} + with tempfile.TemporaryDirectory(prefix='pulse-notification-runtime-') as temporary: + data = Path(temporary) + sink = Sink() + journey = None + try: + with (data / 'owned-server.log').open('wb') as log: + journey = Journey(binary, data, log, sink) + journey.run() + report['outcome'] = 'passed' + except Exception as exc: + report['error'] = str(exc) if isinstance(exc, Failure) else type(exc).__name__ + finally: + try: + sink.close() + report['sink_thread_joined_and_listener_closed'] = True + except Exception as exc: + report['sink_thread_joined_and_listener_closed'] = False + report['cleanup_error'] = type(exc).__name__ + report['outcome'] = 'failed' + if journey: + report.update(phases=journey.phases, child_exits=journey.exits, + requests=journey.requests, incomplete_request=journey.active_request, + deliveries=journey.deliveries, delivery_observations=journey.delivery_observations) + report['owned_temporary_data_removed'] = not data.exists() + report['source_clean_after'] = subprocess.run(['git', 'diff', '--quiet', 'HEAD'], check=False).returncode == 0 + if not report['source_clean_after']: + report['outcome'] = 'failed' + if args.receipt: + # Never replace a previous attempt's receipt. No logs or credentials. + with args.receipt.open('x', encoding='utf-8') as handle: + json.dump(report, handle, sort_keys=True, indent=2) + handle.write('\n') + print('NOTIFICATION_RUNTIME=' + json.dumps(report, sort_keys=True), flush=True) + return 0 if report['outcome'] == 'passed' else 1 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/scripts/tests/test_notification_runtime.py b/scripts/tests/test_notification_runtime.py new file mode 100644 index 0000000000..7d9ad138d4 --- /dev/null +++ b/scripts/tests/test_notification_runtime.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""Controls for the source-bound ordinary notification journey, without a server.""" +import contextlib +import copy +import importlib.util +import io +from pathlib import Path +from types import SimpleNamespace +import unittest +from unittest import mock + +ROOT = Path(__file__).resolve().parents[2] +SPEC = importlib.util.spec_from_file_location('notification_runtime', ROOT / 'scripts/check_notification_runtime.py') +RUNTIME = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(RUNTIME) +SHA = 'a' * 40 + + +class FakeResponse: + def __init__(self, raw=b'{}', status=200): + self.raw, self.status = raw, status + self.limit = None + + def __enter__(self): + return self + + def __exit__(self, *_): + return None + + def read(self, limit): + self.limit = limit + return self.raw + + +class NotificationRuntimeControls(unittest.TestCase): + def valid_row(self): + return {'path': '/owned-alert', 'body': {'resourceName': 'owned.example.test', + 'instance': RUNTIME.PUBLIC_URL, 'event': 'alert', 'type': 'cpu'}} + + def journey(self): + return RUNTIME.Journey(Path('/unused-binary'), Path('/unused-data'), io.BytesIO(), None) + + def test_current_ordinary_event_not_resolution_or_invented_firing(self): + source = (ROOT / 'internal/notifications/notifications.go').read_text() + prepare = source.split('func (n *NotificationManager) prepareWebhookData(', 1)[1] + self.assertRegex(prepare, r'Event:\s+"alert",') + self.assertEqual(RUNTIME.ORDINARY_EVENT, 'alert') + row = self.valid_row() + self.assertTrue(all(RUNTIME.delivery_checks(row, 'owned.example.test').values())) + for event in ('firing', 'resolved', None): + with self.subTest(event=event): + row['body']['event'] = event + self.assertFalse(RUNTIME.delivery_checks(row, 'owned.example.test')['ordinary_event']) + + def test_all_five_delivery_conditions_are_independent(self): + cases = [('path', '/foreign', 'owned_path'), + ('resourceName', 'other.example.test', 'resource_identity'), + ('instance', 'https://other.example.test', 'saved_public_url'), + ('event', 'resolved', 'ordinary_event'), + ('type', 'memory', 'cpu_metric')] + for field, value, failing in cases: + with self.subTest(field=field): + row = copy.deepcopy(self.valid_row()) + target = row if field == 'path' else row['body'] + target[field] = value + checks = RUNTIME.delivery_checks(row, 'owned.example.test') + self.assertEqual([key for key, passed in checks.items() if not passed], [failing]) + + def test_uncertain_mutation_is_never_replayed_and_keeps_request_identity(self): + journey = self.journey() + client = mock.Mock() + client.open.side_effect = TimeoutError('private provider text must not be printed') + journey.client = client + with contextlib.redirect_stdout(io.StringIO()) as output: + with self.assertRaises(TimeoutError): + journey.request('/api/alerts/config', {'privateValue': 'do-not-log'}, method='PUT') + client.open.assert_called_once() + self.assertEqual(journey.active_request, {'path': '/api/alerts/config', 'method': 'PUT', + 'authentication': 'session'}) + self.assertEqual(journey.requests, []) + self.assertNotIn('do-not-log', output.getvalue()) + self.assertNotIn('private provider text', output.getvalue()) + + def test_expected_refusal_does_not_decode_or_print_response_body(self): + journey = self.journey() + response = FakeResponse(b'private rejection detail', 403) + journey.client = mock.Mock() + journey.client.open.return_value = response + with contextlib.redirect_stdout(io.StringIO()) as output: + self.assertIsNone(journey.request('/api/system/settings', expected=403)) + self.assertEqual(response.limit, RUNTIME.LIMIT + 1) + self.assertNotIn('private rejection detail', output.getvalue()) + self.assertEqual(journey.requests[-1]['status'], 403) + self.assertIsNone(journey.active_request) + + def test_oversized_success_stops_after_one_read(self): + journey = self.journey() + response = FakeResponse(b'x' * (RUNTIME.LIMIT + 1)) + journey.client = mock.Mock() + journey.client.open.return_value = response + with contextlib.redirect_stdout(io.StringIO()): + with self.assertRaisesRegex(RUNTIME.Failure, 'byte bound'): + journey.request('/api/resources') + journey.client.open.assert_called_once() + + def test_only_clean_matching_checkout_and_binary_admitted(self): + valid = '\tbuild\tvcs=git\n\tbuild\tvcs.revision=' + SHA + '\n\tbuild\tvcs.modified=false\n' + cases = [(SHA, 0, valid, True), + ('b' * 40, 0, valid, False), + (SHA, 1, valid, False), + (SHA, 0, valid.replace('modified=false', 'modified=true'), False), + (SHA, 0, valid.replace(SHA, 'b' * 40), False), + (SHA, 0, valid.replace('vcs=git', 'vcs=hg'), False), + (SHA, 0, '', False)] + for actual, dirty, binary, allowed in cases: + with self.subTest(actual=actual, dirty=dirty, binary=binary): + with mock.patch.object(RUNTIME.subprocess, 'check_output', side_effect=[actual, binary]), \ + mock.patch.object(RUNTIME.subprocess, 'run', return_value=SimpleNamespace(returncode=dirty)): + if allowed: + self.assertEqual(RUNTIME.source_binding(Path('/owned-binary'), SHA)['checkout_sha'], SHA) + else: + with self.assertRaises(RUNTIME.Failure): + RUNTIME.source_binding(Path('/owned-binary'), SHA) + + def test_malformed_source_identity_stops_before_commands(self): + with mock.patch.object(RUNTIME.subprocess, 'check_output') as command: + with self.assertRaises(RUNTIME.Failure): + RUNTIME.source_binding(Path('/owned-binary'), 'main') + command.assert_not_called() + + def test_owned_child_has_no_ambient_credentials_or_telemetry(self): + journey = self.journey() + child = mock.Mock() + child.poll.return_value = None + journey.request = mock.Mock(return_value=None) + with mock.patch.dict(RUNTIME.os.environ, {'PATH': '/usr/bin', 'AMBIENT_SECRET': 'not-forwarded'}, clear=True), \ + mock.patch.object(RUNTIME, 'free_port', return_value=7655), \ + mock.patch.object(RUNTIME.subprocess, 'Popen', return_value=child) as launch: + journey.start() + env = launch.call_args.kwargs['env'] + self.assertNotIn('AMBIENT_SECRET', env) + self.assertEqual(env['PULSE_TELEMETRY'], 'false') + self.assertEqual(env['PULSE_MOCK_MODE'], 'false') + self.assertEqual(env['DISCOVERY_ENABLED'], 'false') + self.assertEqual(env['BIND_ADDRESS'], '127.0.0.1') + self.assertTrue(launch.call_args.kwargs['start_new_session']) + + def test_ci_reuses_the_real_embed_with_exact_source_and_attempt_receipt(self): + workflow = (ROOT / '.github/workflows/build-and-test.yml').read_text() + section = workflow.split('- name: Check ordinary notification policy and restart', 1)[1] + section = section.split('# An audit service failure', 1)[0] + self.assertIn('timeout-minutes: 3', section) + self.assertIn('--binary "$RUNNER_TEMP/pulse-real-embed"', section) + self.assertIn('--expected-source-sha "${{ github.sha }}"', section) + self.assertIn('--receipt "$RUNNER_TEMP/pulse-notification-runtime.json"', section) + self.assertIn('if: always()', section) + self.assertIn('notification-runtime-${{ github.run_id }}-${{ github.run_attempt }}', section) + self.assertNotIn('sudo', section) + self.assertNotIn('go build', section) + + +if __name__ == '__main__': + unittest.main(verbosity=2) From 89a3381ffc5a565efdd8c3ba279abb5fe85b7a59 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 11 Oct 2026 05:18:21 +0100 Subject: [PATCH 2/2] Respect deployment telemetry lock in notification CI journey Keep outbound telemetry disabled through the process environment, test the expected locked-field refusal, and save only the unrelated notification fields. Assert the lock on restart rather than attempting to persist it. Contract-Neutral: Correct the CI-only synthetic journey and its controls; product settings, deployment restrictions, permissions and runtime contracts are unchanged. Change-source: pulse-maintainer --- scripts/check_notification_runtime.py | 14 ++++++++-- scripts/tests/test_notification_runtime.py | 31 ++++++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/scripts/check_notification_runtime.py b/scripts/check_notification_runtime.py index 4c7fe197f3..3efe994f71 100644 --- a/scripts/check_notification_runtime.py +++ b/scripts/check_notification_runtime.py @@ -225,10 +225,21 @@ class Journey: {'name': 'unlisted-loopback', 'url': f'http://127.0.0.2:{self.sink.port}/blocked', 'enabled': True, 'method': 'POST', 'service': 'generic'}, expected=400) + def save_system_policy(self): + # This process's deployment owns telemetry=false. Do not weaken that + # lock or include its field in an otherwise unrelated settings patch. + self.request('/api/system/settings/update', {'telemetryEnabled': False}, expected=409) + self.phase('deployment-telemetry-lock-preserved') + self.request('/api/system/settings/update', {'publicURL': PUBLIC_URL, + 'webhookAllowedPrivateCIDRs': '127.0.0.1/32'}) + def assert_settings(self): settings = self.request('/api/system/settings') require(settings.get('publicURL') == PUBLIC_URL and settings.get('webhookAllowedPrivateCIDRs') == '127.0.0.1/32', 'saved system policy changed') + require(settings.get('telemetryEnabled') is False and + settings.get('envOverrides', {}).get('PULSE_TELEMETRY') is True, + 'deployment telemetry restriction changed') cfg = self.request('/api/alerts/config') require(cfg.get('enabled') and cfg.get('activationState') == 'active' and cfg.get('agentDefaults', {}).get('cpu', {}).get('trigger') == 80, @@ -277,8 +288,7 @@ class Journey: require(any(c.name == 'pulse_csrf' for c in self.cookies), 'login missing CSRF cookie') require(self.request('/api/resources?limit=500').get('data') == [], 'initial live estate not empty') self.phase('authenticated-empty-live-startup') - self.request('/api/system/settings/update', {'publicURL': PUBLIC_URL, - 'webhookAllowedPrivateCIDRs': '127.0.0.1/32', 'telemetryEnabled': False}) + self.save_system_policy() self.denied_destination() self.request('/api/notifications/webhooks', {'name': 'owned-first-alert', 'url': f'http://127.0.0.1:{self.sink.port}/owned-alert', 'enabled': True, diff --git a/scripts/tests/test_notification_runtime.py b/scripts/tests/test_notification_runtime.py index 7d9ad138d4..3eb2f7ac1e 100644 --- a/scripts/tests/test_notification_runtime.py +++ b/scripts/tests/test_notification_runtime.py @@ -103,6 +103,37 @@ class NotificationRuntimeControls(unittest.TestCase): journey.request('/api/resources') journey.client.open.assert_called_once() + def test_settings_patch_preserves_and_tests_deployment_telemetry_lock(self): + journey = self.journey() + journey.request = mock.Mock() + with contextlib.redirect_stdout(io.StringIO()): + journey.save_system_policy() + self.assertEqual(journey.request.call_args_list, [ + mock.call('/api/system/settings/update', {'telemetryEnabled': False}, expected=409), + mock.call('/api/system/settings/update', {'publicURL': RUNTIME.PUBLIC_URL, + 'webhookAllowedPrivateCIDRs': '127.0.0.1/32'})]) + self.assertEqual(journey.phases, ['deployment-telemetry-lock-preserved']) + handler = (ROOT / 'internal/api/system_settings.go').read_text() + self.assertIn('rawRequest["telemetryEnabled"]; ok && h.config.EnvOverrides["PULSE_TELEMETRY"]', handler) + + def test_policy_readback_requires_disabled_locked_telemetry(self): + system = {'publicURL': RUNTIME.PUBLIC_URL, 'webhookAllowedPrivateCIDRs': '127.0.0.1/32', + 'telemetryEnabled': False, 'envOverrides': {'PULSE_TELEMETRY': True}} + alerts = {'enabled': True, 'activationState': 'active', + 'agentDefaults': {'cpu': {'trigger': 80}}} + for telemetry, lock, allowed in [(False, True, True), (True, True, False), + (None, True, False), (False, False, False)]: + with self.subTest(telemetry=telemetry, lock=lock): + journey = self.journey() + settings = copy.deepcopy(system) + settings['telemetryEnabled'], settings['envOverrides']['PULSE_TELEMETRY'] = telemetry, lock + journey.request = mock.Mock(side_effect=[settings, alerts]) + if allowed: + journey.assert_settings() + else: + with self.assertRaisesRegex(RUNTIME.Failure, 'telemetry restriction'): + journey.assert_settings() + def test_only_clean_matching_checkout_and_binary_admitted(self): valid = '\tbuild\tvcs=git\n\tbuild\tvcs.revision=' + SHA + '\n\tbuild\tvcs.modified=false\n' cases = [(SHA, 0, valid, True),