Bind demo cohost observation to the real Relay health route

Current operational source exposes healthz and takes the shared deployment lock. Pin that route and make the disposable service return 404 for unknown paths so native acceptance cannot manufacture a healthy cohost observation.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-03 01:44:42 +01:00
parent 4df3e8974e
commit 360721cc91
4 changed files with 13 additions and 4 deletions
+6 -3
View File
@@ -38,11 +38,14 @@ def fail():
if %r: threading.Thread(target=fail,daemon=True).start()
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
expected = {'/healthz'} if %r == 17656 else {'/api/health','/api/version'}
if self.path not in expected:
self.send_response(404); self.end_headers(); return
body=json.dumps({'version':VERSION} if self.path=='/api/version' else {'status':'healthy'}).encode()
self.send_response(200); self.end_headers(); self.wfile.write(body)
def log_message(self,*args): pass
http.server.HTTPServer(('127.0.0.1',%r),Handler).serve_forever()
''' % (version, fail_after, fail_after, port)).encode()
''' % (version, fail_after, fail_after, port, port)).encode()
def unit(executable, user="pulse"):
@@ -128,9 +131,9 @@ def main():
original = (SCRIPTS / "demo-runtime-transaction.py").read_bytes()
# Only the cohost endpoint is rebound. Transaction, paths, locking,
# systemd, HTTP/TLS, estate, clocks and watches remain real and unchanged.
endpoint = b'https://relay.pulserelay.pro/health'
endpoint = b'https://relay.pulserelay.pro/healthz'
assert original.count(endpoint) == 1
source = original.replace(endpoint, b'https://127.0.0.1:18444/health')
source = original.replace(endpoint, b'https://127.0.0.1:18444/healthz')
result["original_engine_sha256"] = hashlib.sha256(original).hexdigest()
result["fixture_engine_sha256"] = hashlib.sha256(source).hexdigest()
result["bootstrap_sha256"] = hashlib.sha256(dispatcher.BOOTSTRAP.encode()).hexdigest()
@@ -347,6 +347,9 @@ class TransactionTest(unittest.TestCase):
class InputAndCommandTest(unittest.TestCase):
def test_cohost_check_uses_the_existing_relay_health_route(self):
self.assertEqual(engine.RELAY_HEALTH, "https://relay.pulserelay.pro/healthz")
def test_valid_current_envelope(self):
engine.validate(request())
engine.validate(request("update"))