diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts index ad5d984426..b2392b6b5b 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts @@ -1494,6 +1494,62 @@ test('waits passively for a Host that cannot be taken over', async () => { await owner.close(); }); +test('offers to stop an exact local ephemeral Host before retrying startup', async () => { + const observed = incompatibleHost('blocked_by_residency'); + const conflict = { + ...observed, + registration: { ...observed.registration, lifecycleMode: 'ephemeral' as const }, + processIdentity: { + startIdentity: 'darwin:1700000000:123456', + }, + handshake: { + ...observed.handshake, + activity: { + connections: 0, + activeOperations: 0, + processUptimeSeconds: 60, + residencies: [], + }, + }, + }; + const replacement = candidateHarness(); + let starts = 0; + let prompts = 0; + let terminations = 0; + const owner = await startRuntimeHostDesktopManager( + { rootPath: '/workspace' } as DesktopRuntimeHostCandidateStartInput, + { + startCandidate: async () => { + starts += 1; + return starts === 1 ? conflict : ready(replacement.candidate); + }, + upgradePrompts: { + restartable: async () => assert.fail('incompatible Host used restart prompt'), + nonRestartable: async (_conflict, action) => { + prompts += 1; + assert.equal(action, 'replace_may_interrupt_work'); + return 'replace'; + }, + }, + forceTerminateObservedHost: async (identity, authority) => { + terminations += 1; + assert.deepEqual(identity, { + rootPath: '/workspace', + registration: conflict.registration, + }); + assert.deepEqual(authority.processIdentity, conflict.processIdentity); + assert.equal(authority.isCurrent(), true); + return true; + }, + }, + ); + + assert.equal(prompts, 1, 'even an idle snapshot must not authorize a forced stop'); + assert.equal(terminations, 1); + assert.equal(starts, 2); + await owner.close(); +}); + test('silently replaces an idle non-restartable Local Host and retries', async () => { const observed = upgradeRequired(true); const conflict = { @@ -1667,7 +1723,7 @@ test('lets the user cancel startup when an incompatible Host owns the root', asy function incompatibleHost( replacement: 'wait_for_idle_exit' | 'blocked_by_residency', -): DesktopRuntimeHostCandidateStartResult { +): Extract { return { kind: 'incompatible', registration: hostRegistration({ compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH - 1 }), diff --git a/apps/desktop/src/main/__tests__/runtime-host-upgrade-dialog.test.ts b/apps/desktop/src/main/__tests__/runtime-host-upgrade-dialog.test.ts index 2c4abd86c1..9eea55398e 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-upgrade-dialog.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-upgrade-dialog.test.ts @@ -73,9 +73,9 @@ test('maps the non-default replacement choice to the replace decision', async () const prompts = createRuntimeHostUpgradePrompts( async () => 'en', async (options) => { - assert.deepEqual(options.buttons, ['Stop Host and Continue', 'Cancel Startup']); - assert.equal(options.defaultId, 1); - assert.equal(options.cancelId, 1); + assert.deepEqual(options.buttons, ['Stop Host and Continue', 'Wait', 'Cancel Startup']); + assert.equal(options.defaultId, 2); + assert.equal(options.cancelId, 2); assert.match(options.detail ?? '', /Maka will stop this Host/); return { response: 0, checkboxChecked: false }; }, @@ -85,7 +85,7 @@ test('maps the non-default replacement choice to the replace decision', async () { kind: 'upgrade_required', restartable: false, - registration: { pid: 42 }, + registration: { pid: 42, lifecycleMode: 'ephemeral' }, } as never, 'replace_may_interrupt_work', ), diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts index 6731a4aae2..36d8eca62d 100644 --- a/apps/desktop/src/main/runtime-host-desktop-manager.ts +++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts @@ -21,6 +21,7 @@ import { randomUUID } from 'node:crypto'; import type { BotIncomingMessage } from '@maka/runtime/bots'; import { abortable, + forceTerminateObservedRegisteredRuntimeHost, forceTerminateRegisteredRuntimeHost, RuntimeHostOperationError, RuntimeHostPermanentReconnectError, @@ -247,6 +248,7 @@ export async function startRuntimeHostDesktopManager( upgradePrompts?: RuntimeHostUpgradePrompts; waitForHostExit?: (pid: number) => Promise; forceTerminateHost?: typeof forceTerminateRegisteredRuntimeHost; + forceTerminateObservedHost?: typeof forceTerminateObservedRegisteredRuntimeHost; waitForHostRetirement?: ( registration: HostRegistration, signal: AbortSignal, @@ -271,6 +273,7 @@ export async function startRuntimeHostDesktopManager( options.upgradePrompts, options.waitForHostExit ?? waitForProcessExit, options.forceTerminateHost ?? forceTerminateRegisteredRuntimeHost, + options.forceTerminateObservedHost ?? forceTerminateObservedRegisteredRuntimeHost, options.waitForHostRetirement ?? waitForProcessRetirement, options.resolveLocalHostReplacement, options.recoverLocalHost, @@ -310,6 +313,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { private readonly upgradePrompts: RuntimeHostUpgradePrompts | undefined, private readonly waitForHostExit: (pid: number) => Promise, private readonly forceTerminateHost: typeof forceTerminateRegisteredRuntimeHost, + private readonly forceTerminateObservedHost: typeof forceTerminateObservedRegisteredRuntimeHost, private readonly waitForHostRetirement: ( registration: HostRegistration, signal: AbortSignal, @@ -1138,7 +1142,8 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { ) { const replacement = target.input.profileTarget ? undefined - : await this.resolveLocalHostReplacement?.(result.registration, signal); + : this.#registeredEphemeralHostReplacement(target, result, signal) ?? + (await this.resolveLocalHostReplacement?.(result.registration, signal)); const activity = result.handshake?.activity; if (replacement && activity && isHostActivityIdle(activity)) { // Only a complete, observed snapshot can authorize silent @@ -1184,6 +1189,43 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { } } + #registeredEphemeralHostReplacement( + target: DesktopRuntimeHostTargetGeneration, + conflict: RuntimeHostWaitConflict, + signal: AbortSignal, + ): RuntimeHostLocalReplacement | undefined { + const { registration, processIdentity } = conflict; + if (registration.lifecycleMode !== 'ephemeral' || !processIdentity) return undefined; + const stillAuthorized = () => + !this.#closed && + !signal.aborted && + target.valid && + this.#targets.get(target.target.profile.id) === target; + return { + replace: async (activeWorkPolicy) => { + // This Host cannot participate in the current retirement protocol, so + // an earlier idle snapshot cannot prove that it remains idle. Require + // explicit consent before using the identity-fenced termination path. + if (activeWorkPolicy === 'refuse_active_work') return 'active_tasks'; + signal.throwIfAborted(); + const terminated = await this.forceTerminateObservedHost( + { + rootPath: this.#baseInput.rootPath, + registration, + }, + { processIdentity, isCurrent: stillAuthorized }, + ); + signal.throwIfAborted(); + if (!terminated) { + throw new RuntimeHostPermanentReconnectError( + 'The older Runtime Host changed before it could be stopped safely', + ); + } + return 'replaced'; + }, + }; + } + #resolveRestartable( conflict: RuntimeHostRestartableConflict, ): Promise { diff --git a/apps/desktop/src/main/runtime-host-upgrade-copy.ts b/apps/desktop/src/main/runtime-host-upgrade-copy.ts index 3f03574171..e08d2574cd 100644 --- a/apps/desktop/src/main/runtime-host-upgrade-copy.ts +++ b/apps/desktop/src/main/runtime-host-upgrade-copy.ts @@ -59,7 +59,9 @@ export function buildRuntimeHostUpgradeDialog( : undefined; const canWait = availability === 'wait' || - (availability === 'restart' && conflict.registration.lifecycleMode !== 'service'); + (availability === 'restart' && conflict.registration.lifecycleMode !== 'service') || + (availability === 'replace_may_interrupt_work' && + conflict.registration.lifecycleMode === 'ephemeral'); if (action) { choices.push({ label: action === 'restart' ? copy.restart : copy.replace, diff --git a/native/runtime-host-peer/src/lib.rs b/native/runtime-host-peer/src/lib.rs index 4017167406..13ffe3f389 100644 --- a/native/runtime-host-peer/src/lib.rs +++ b/native/runtime-host-peer/src/lib.rs @@ -19,6 +19,7 @@ mod bindings; mod engine; +mod process_identity; mod webrtc_direct; #[cfg(target_os = "windows")] mod windows_lifecycle; @@ -29,6 +30,7 @@ pub use bindings::{ PeerTransitSnapshot, StartPeerEndpointOptions, ensure_peer_identity, sign_peer_identity, start_peer_endpoint, verify_peer_identity, }; +pub use process_identity::read_process_start_identity; #[cfg(target_os = "windows")] pub use windows_lifecycle::{ WindowsTaskStatus, own_current_process_tree, windows_task_activate, windows_task_converge, diff --git a/native/runtime-host-peer/src/process_identity.rs b/native/runtime-host-peer/src/process_identity.rs new file mode 100644 index 0000000000..96016da579 --- /dev/null +++ b/native/runtime-host-peer/src/process_identity.rs @@ -0,0 +1,176 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +use napi_derive::napi; + +/// Returns an opaque identifier for one OS process lifetime. Unlike a command +/// line, this value cannot be forged accidentally by an argument or path. +#[napi] +pub fn read_process_start_identity(pid: u32) -> Option { + if pid == 0 { + return None; + } + read_platform_process_start_identity(pid) +} + +#[cfg(target_os = "linux")] +fn read_platform_process_start_identity(pid: u32) -> Option { + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + let boot_id = std::fs::read_to_string("/proc/sys/kernel/random/boot_id").ok()?; + let boot_id = boot_id.trim(); + if boot_id.is_empty() + || !boot_id + .bytes() + .all(|byte| byte.is_ascii_hexdigit() || byte == b'-') + { + return None; + } + let start_ticks = linux_process_start_ticks(&stat)?; + Some(format!("linux:{boot_id}:{start_ticks}")) +} + +#[cfg(target_os = "linux")] +fn linux_process_start_ticks(stat: &str) -> Option<&str> { + // `comm` (field 2) may contain spaces and parentheses. The final `)` is + // the only safe boundary before field 3; starttime is field 22. + let mut fields = stat.get(stat.rfind(')')? + 1..)?.split_ascii_whitespace(); + let start_ticks = fields.nth(19)?; + (!start_ticks.is_empty() && start_ticks.bytes().all(|byte| byte.is_ascii_digit())) + .then_some(start_ticks) +} + +#[cfg(target_os = "macos")] +fn read_platform_process_start_identity(pid: u32) -> Option { + use std::mem::{MaybeUninit, size_of}; + + const PROC_PIDTBSDINFO: i32 = 3; + + #[repr(C)] + struct ProcBsdInfo { + pbi_flags: u32, + pbi_status: u32, + pbi_xstatus: u32, + pbi_pid: u32, + pbi_ppid: u32, + pbi_uid: u32, + pbi_gid: u32, + pbi_ruid: u32, + pbi_rgid: u32, + pbi_svuid: u32, + pbi_svgid: u32, + rfu_1: u32, + pbi_comm: [u8; 16], + pbi_name: [u8; 32], + pbi_nfiles: u32, + pbi_pgid: u32, + pbi_pjobc: u32, + e_tdev: u32, + e_tpgid: u32, + pbi_nice: i32, + pbi_start_tvsec: u64, + pbi_start_tvusec: u64, + } + + unsafe extern "C" { + fn proc_pidinfo( + pid: i32, + flavor: i32, + arg: u64, + buffer: *mut core::ffi::c_void, + buffer_size: i32, + ) -> i32; + } + + let mut info = MaybeUninit::::zeroed(); + let expected_size = size_of::(); + // SAFETY: `info` points to writable storage of exactly the size passed to + // libproc. The value is read only when libproc reports a complete record. + let read_size = unsafe { + proc_pidinfo( + pid as i32, + PROC_PIDTBSDINFO, + 0, + info.as_mut_ptr().cast(), + expected_size as i32, + ) + }; + if read_size != expected_size as i32 { + return None; + } + // SAFETY: the successful full-size call initialized every byte. + let info = unsafe { info.assume_init() }; + if info.pbi_pid != pid || info.pbi_start_tvsec == 0 || info.pbi_start_tvusec >= 1_000_000 { + return None; + } + Some(format!( + "darwin:{}:{}", + info.pbi_start_tvsec, info.pbi_start_tvusec + )) +} + +#[cfg(target_os = "windows")] +fn read_platform_process_start_identity(pid: u32) -> Option { + use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; + use windows::Win32::{ + Foundation::{FILETIME, HANDLE}, + System::Threading::{GetProcessTimes, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION}, + }; + + // SAFETY: the returned valid handle is immediately placed under RAII. + let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, false, pid) }.ok()?; + // SAFETY: ownership of the newly opened handle is transferred exactly once. + let owned = unsafe { OwnedHandle::from_raw_handle(handle.0) }; + let handle = HANDLE(owned.as_raw_handle()); + let mut created = FILETIME::default(); + let mut exited = FILETIME::default(); + let mut kernel = FILETIME::default(); + let mut user = FILETIME::default(); + // SAFETY: the handle remains live and all four output pointers are valid. + unsafe { GetProcessTimes(handle, &mut created, &mut exited, &mut kernel, &mut user) }.ok()?; + let ticks = (u64::from(created.dwHighDateTime) << 32) | u64::from(created.dwLowDateTime); + (ticks != 0).then(|| format!("windows:{ticks}")) +} + +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] +fn read_platform_process_start_identity(_pid: u32) -> Option { + None +} + +#[cfg(test)] +mod common_tests { + use super::read_process_start_identity; + + #[test] + fn current_process_identity_is_stable() { + let first = read_process_start_identity(std::process::id()); + assert!(first.is_some()); + assert_eq!(read_process_start_identity(std::process::id()), first); + } +} + +#[cfg(all(test, target_os = "linux"))] +mod linux_tests { + use super::linux_process_start_ticks; + + #[test] + fn parses_start_ticks_after_a_hostile_process_name() { + let stat = "42 (workspace --startup-attempt-id 00000000-0000-4000-8000-000000000001) S 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 98765 20"; + assert_eq!(linux_process_start_ticks(stat), Some("98765")); + } +} diff --git a/packages/runtime-host/src/__tests__/handshake-compatibility.test.ts b/packages/runtime-host/src/__tests__/handshake-compatibility.test.ts index a2e93d3f62..fbb9f69c82 100644 --- a/packages/runtime-host/src/__tests__/handshake-compatibility.test.ts +++ b/packages/runtime-host/src/__tests__/handshake-compatibility.test.ts @@ -103,9 +103,39 @@ test('receives structured incompatibility guidance from the released v0.1.11 Hos assert.equal(result.handshake.compatibilityEpoch, V0_1_11_HOST_COMPATIBILITY_EPOCH); assert.equal(result.handshake.compositionRevision, V0_1_11_HOST_REVISION); assert.equal(result.handshake.replacement, 'blocked_by_residency'); + assert.deepEqual(result.processIdentity, { + startIdentity: 'darwin:1700000000:123456', + }); } }, - { registrationCompatibilityEpoch: V0_1_11_HOST_COMPATIBILITY_EPOCH }, + { + registrationCompatibilityEpoch: V0_1_11_HOST_COMPATIBILITY_EPOCH, + registrationLifecycleMode: 'ephemeral', + readProcessIdentity: async () => ({ + startIdentity: 'darwin:1700000000:123456', + }), + }, + ); +}); + +test('process identity query failure does not consume or prevent the incompatible handshake', async () => { + await withForgedHandshakePeer( + async (transport, hostEpoch, rootId) => { + const rawHello = await transport.read(2_000); + await admitV0_1_11ClientHello({ rawHello, transport, hostEpoch, rootId }); + await transport.closed; + }, + async (result) => { + assert.equal(result.kind, 'incompatible'); + if (result.kind === 'incompatible') assert.equal(result.processIdentity, undefined); + }, + { + registrationCompatibilityEpoch: V0_1_11_HOST_COMPATIBILITY_EPOCH, + registrationLifecycleMode: 'ephemeral', + readProcessIdentity: async () => { + throw new Error('process identity unavailable'); + }, + }, ); }); @@ -326,8 +356,10 @@ async function withForgedHandshakePeer( options: { readonly registrationCompatibilityEpoch?: number; readonly registrationRootId?: string; + readonly registrationLifecycleMode?: 'ephemeral'; readonly expectConnection?: boolean; readonly prepareAfterListen?: boolean; + readonly readProcessIdentity?: () => Promise<{ readonly startIdentity: string } | undefined>; } = {}, ): Promise { const base = await mkdtemp(join(tmpdir(), 'maka-runtime-host-handshake-')); @@ -358,7 +390,9 @@ async function withForgedHandshakePeer( hostEpoch, }); const serverTask = deferred(); + let endpointConnected = false; const server = createServer((socket) => { + endpointConnected = true; void serve(new FramedTransport(socket), hostEpoch, capability.rootId).then( serverTask.resolve, serverTask.reject, @@ -382,14 +416,30 @@ async function withForgedHandshakePeer( compositionId: 'maka.interactive', compositionRevision: '1', state: 'ready', + ...(options.registrationLifecycleMode + ? { lifecycleMode: options.registrationLifecycleMode } + : {}), pid: process.pid, createdAt: new Date().toISOString(), }); + const readProcessIdentity = options.readProcessIdentity; const resolved = await connectResolvedRuntimeHost({ capability, controlDirectory, clientInstanceId: randomUUID(), protocol: PROTOCOL, + ...(readProcessIdentity + ? { + readProcessIdentity: async () => { + assert.equal( + endpointConnected, + false, + 'process identity must be observed before opening the Host endpoint', + ); + return readProcessIdentity(); + }, + } + : {}), }); if (resolved.kind === 'election_deadline_elapsed') { throw new Error('Unexpected Runtime Host election deadline'); diff --git a/packages/runtime-host/src/__tests__/process-identity.test.ts b/packages/runtime-host/src/__tests__/process-identity.test.ts new file mode 100644 index 0000000000..11ddd7b873 --- /dev/null +++ b/packages/runtime-host/src/__tests__/process-identity.test.ts @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { linuxProcessStartTicks } from '../client/process-identity.js'; + +test('reads Linux process start time without interpreting hostile command text', () => { + const stat = + '42 (workspace --startup-attempt-id 00000000-0000-4000-8000-000000000001) S 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 98765 20'; + assert.equal(linuxProcessStartTicks(stat), '98765'); + assert.equal(linuxProcessStartTicks('42 malformed'), undefined); +}); diff --git a/packages/runtime-host/src/__tests__/registered-host-termination.test.ts b/packages/runtime-host/src/__tests__/registered-host-termination.test.ts index 98f073231b..cd95dce806 100644 --- a/packages/runtime-host/src/__tests__/registered-host-termination.test.ts +++ b/packages/runtime-host/src/__tests__/registered-host-termination.test.ts @@ -26,8 +26,11 @@ import { prepareStorageRootControlDirectory, resolveStorageRoot, } from '@maka/storage/root-authority'; -import { forceTerminateRegisteredRuntimeHostWithDependencies } from '../client/registered-host-termination.js'; -import { writeHostRegistration } from '../control/registration.js'; +import { + forceTerminateObservedRegisteredRuntimeHostWithDependencies, + forceTerminateRegisteredRuntimeHostWithDependencies, +} from '../client/registered-host-termination.js'; +import { readHostRegistration, writeHostRegistration } from '../control/registration.js'; import { RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, @@ -35,7 +38,7 @@ import { type HostRegistration, } from '../protocol/index.js'; -test('forced termination remains bound to the registered Host identity', async (t) => { +test('owned forced termination remains bound to the registered Host identity', async (t) => { const rootPath = await mkdtemp(join(tmpdir(), 'maka-host-termination-')); t.after(() => rm(rootPath, { recursive: true, force: true })); const capability = await resolveStorageRoot({ path: rootPath, kind: 'interactive' }); @@ -119,3 +122,106 @@ test('forced termination remains bound to the registered Host identity', async ( ); assert.equal(terminated, 1); }); + +test('observed forced termination remains bound to the exact process instance', async (t) => { + const rootPath = await mkdtemp(join(tmpdir(), 'maka-host-termination-')); + t.after(() => rm(rootPath, { recursive: true, force: true })); + const capability = await resolveStorageRoot({ path: rootPath, kind: 'interactive' }); + const { controlDirectory } = await prepareStorageRootControlDirectory(capability); + const identity = { + rootPath, + rootId: capability.rootId, + hostEpoch: 'expected-epoch', + pid: 4242, + }; + const registration: HostRegistration = { + kind: 'maka-runtime-host', + schemaVersion: RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION, + rootId: capability.rootId, + hostEpoch: identity.hostEpoch, + endpoint: join(rootPath, 'runtime-host.sock'), + protocolMin: RUNTIME_HOST_PROTOCOL_VERSION, + protocolMax: RUNTIME_HOST_PROTOCOL_VERSION, + compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH, + compositionId: 'maka.interactive', + compositionRevision: 'test', + lifecycleMode: 'ephemeral', + state: 'ready', + pid: identity.pid, + createdAt: new Date(0).toISOString(), + }; + const processIdentity = { + startIdentity: 'darwin:1700000000:123456', + }; + let alive = true; + let terminated = 0; + let replaceBeforeSignal = false; + let replaceCreatedAtBeforeSignal = false; + let replaceProcessBeforeSignal = false; + let stillAuthorized = true; + let releaseOwnershipBeforeSignal = false; + const dependencies = { + isProcessAlive: () => alive, + readProcessIdentity: async () => { + if (releaseOwnershipBeforeSignal) stillAuthorized = false; + return replaceProcessBeforeSignal + ? { startIdentity: 'darwin:1700000001:123456' } + : processIdentity; + }, + readRegistration: async (directory: string) => { + if (replaceBeforeSignal) { + await writeHostRegistration(controlDirectory, { ...registration, hostEpoch: 'successor' }); + } + if (replaceCreatedAtBeforeSignal) { + await writeHostRegistration(controlDirectory, { + ...registration, + createdAt: new Date(1).toISOString(), + }); + } + return readHostRegistration(directory); + }, + settleMs: 0, + signalProcess: () => { + terminated += 1; + alive = false; + return true; + }, + }; + const authority = () => ({ + processIdentity, + isCurrent: () => stillAuthorized, + }); + const terminate = () => + forceTerminateObservedRegisteredRuntimeHostWithDependencies( + { rootPath, registration }, + authority(), + dependencies, + ); + + await writeHostRegistration(controlDirectory, registration); + replaceBeforeSignal = true; + assert.equal(await terminate(), false); + assert.equal(terminated, 0); + + await writeHostRegistration(controlDirectory, registration); + replaceBeforeSignal = false; + replaceCreatedAtBeforeSignal = true; + assert.equal(await terminate(), false); + assert.equal(terminated, 0); + + await writeHostRegistration(controlDirectory, registration); + replaceCreatedAtBeforeSignal = false; + releaseOwnershipBeforeSignal = true; + assert.equal(await terminate(), false); + assert.equal(terminated, 0); + + stillAuthorized = true; + releaseOwnershipBeforeSignal = false; + replaceProcessBeforeSignal = true; + assert.equal(await terminate(), false); + assert.equal(terminated, 0); + + replaceProcessBeforeSignal = false; + assert.equal(await terminate(), true); + assert.equal(terminated, 1); +}); diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts index abbd00d4f5..f9b9e85238 100644 --- a/packages/runtime-host/src/client/connection.ts +++ b/packages/runtime-host/src/client/connection.ts @@ -81,6 +81,10 @@ import { } from './session-subscription.js'; import { ClientCapabilityChannel } from './client-capability-channel.js'; import type { ClientCapabilityProvider } from './client-capability.js'; +import { + readRuntimeHostProcessIdentity, + type RuntimeHostProcessIdentity, +} from './process-identity.js'; const DEFAULT_CONNECT_TIMEOUT_MS = 500; const DEFAULT_HANDSHAKE_TIMEOUT_MS = 2_000; @@ -138,18 +142,21 @@ export type ConnectRuntimeHostResult = kind: 'incompatible'; handshake: HostIncompatible; registration: HostRegistration; + processIdentity?: RuntimeHostProcessIdentity; } | { kind: 'upgrade_required'; registration: HostRegistration; restartable: true; handshake: HostIncompatible; + processIdentity?: RuntimeHostProcessIdentity; } | { kind: 'upgrade_required'; registration: HostRegistration; restartable: false; handshake?: HostIncompatible; + processIdentity?: RuntimeHostProcessIdentity; } | { kind: 'draining'; registration: HostRegistration } | { @@ -234,6 +241,7 @@ interface ConnectResolvedRuntimeHostInput clientInstanceId: string; controlDirectory: string; electionDeadline?: number; + readProcessIdentity?: typeof readRuntimeHostProcessIdentity; } export interface RuntimeHostConnection { @@ -1223,6 +1231,16 @@ export async function connectResolvedRuntimeHost( registration, }; } + // Observe the candidate before opening its endpoint. Besides keeping this + // potentially slow OS query outside the Host's handshake window, the later + // root/epoch-validated handshake binds this evidence to the registration we + // actually reached. Query failure deliberately leaves recovery unavailable. + const processIdentity = shouldObserveProcessIdentity(registration, generation, input.protocol) + ? await (input.readProcessIdentity ?? readRuntimeHostProcessIdentity)(registration.pid).catch( + () => undefined, + ) + : undefined; + const processEvidence = processIdentity === undefined ? {} : { processIdentity }; const connectDeadline = phaseDeadline(connectTimeoutMs, input.electionDeadline); const connectBudget = remainingTimeout(connectDeadline.at); if (connectBudget === undefined) { @@ -1310,7 +1328,12 @@ export async function connectResolvedRuntimeHost( registration.generation !== generation ) { await result.connection.close().catch(() => undefined); - return { kind: 'upgrade_required', registration, restartable: false }; + return { + kind: 'upgrade_required', + registration, + restartable: false, + ...processEvidence, + }; } return { ...result, registration }; } @@ -1331,16 +1354,18 @@ export async function connectResolvedRuntimeHost( registration, restartable: true, handshake: result.handshake, + ...processEvidence, } : { kind: 'upgrade_required', registration, restartable: false, handshake: result.handshake, + ...processEvidence, }; } return result.kind === 'incompatible' - ? { ...result, registration } + ? { ...result, registration, ...processEvidence } : { kind: 'draining', registration }; } catch (error) { transport.abort(); @@ -1383,6 +1408,20 @@ export async function connectResolvedRuntimeHost( } } +function shouldObserveProcessIdentity( + registration: HostRegistration, + generation: string | undefined, + protocol: ProtocolRange, +): boolean { + return ( + registration.lifecycleMode === 'ephemeral' && + (registration.compatibilityEpoch !== RUNTIME_HOST_COMPATIBILITY_EPOCH || + registration.protocolMax < protocol.min || + registration.protocolMin > protocol.max || + (generation !== undefined && registration.generation !== generation)) + ); +} + interface ExchangeRuntimeHostHandshakeInput { readonly transport: RuntimeHostMessageTransport; readonly protocol: ProtocolRange; diff --git a/packages/runtime-host/src/client/index.ts b/packages/runtime-host/src/client/index.ts index 134aede060..35a6c05746 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -45,9 +45,13 @@ export { type RuntimeHostRetirementPreparation, } from './host-retirement.js'; export { + forceTerminateObservedRegisteredRuntimeHost, forceTerminateRegisteredRuntimeHost, + type ObservedRegisteredRuntimeHost, + type ObservedRegisteredRuntimeHostTerminationAuthority, type RegisteredRuntimeHostIdentity, } from './registered-host-termination.js'; +export type { RuntimeHostProcessIdentity } from './process-identity.js'; export { LOCAL_RUNTIME_HOST_PROFILE, RUNTIME_HOST_ACCESS_CREDENTIAL_MAX_BYTES, diff --git a/packages/runtime-host/src/client/process-identity.ts b/packages/runtime-host/src/client/process-identity.ts new file mode 100644 index 0000000000..66d284029d --- /dev/null +++ b/packages/runtime-host/src/client/process-identity.ts @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { readFile } from 'node:fs/promises'; +import { readRuntimeHostNativeProcessStartIdentity } from '../transport/peer-native.js'; + +const PROCESS_QUERY_TIMEOUT_MS = 5_000; + +export interface RuntimeHostProcessIdentity { + readonly startIdentity: string; +} + +/** + * Reads an OS-owned identifier for one process lifetime. Absence and every + * query failure deliberately fail closed. + */ +export async function readRuntimeHostProcessIdentity( + pid: number, +): Promise { + if (!Number.isSafeInteger(pid) || pid <= 0) return undefined; + const nativePath = process.env.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH?.trim(); + if (nativePath) { + try { + const startIdentity = readRuntimeHostNativeProcessStartIdentity(nativePath, pid); + if (startIdentity) return { startIdentity }; + } catch { + // Linux can still use procfs; every other platform fails closed below. + } + } + if (process.platform === 'linux') { + try { + const [stat, bootIdText] = await Promise.all([ + readFile(`/proc/${pid}/stat`, { + encoding: 'utf8', + signal: AbortSignal.timeout(PROCESS_QUERY_TIMEOUT_MS), + }), + readFile('/proc/sys/kernel/random/boot_id', { + encoding: 'utf8', + signal: AbortSignal.timeout(PROCESS_QUERY_TIMEOUT_MS), + }), + ]); + const startTicks = linuxProcessStartTicks(stat); + const bootId = bootIdText.trim(); + if (startTicks && bootId.length > 0 && /^[0-9a-f-]+$/iu.test(bootId)) { + return { startIdentity: `linux:${bootId}:${startTicks}` }; + } + } catch { + // Query failure deliberately leaves destructive recovery unavailable. + } + } + return undefined; +} + +export function linuxProcessStartTicks(stat: string): string | undefined { + const commandEnd = stat.lastIndexOf(')'); + if (commandEnd < 0) return undefined; + const startTicks = stat + .slice(commandEnd + 1) + .trim() + .split(/\s+/u)[19]; + return startTicks && /^\d+$/u.test(startTicks) ? startTicks : undefined; +} diff --git a/packages/runtime-host/src/client/registered-host-termination.ts b/packages/runtime-host/src/client/registered-host-termination.ts index 4832b08c99..0025c80f69 100644 --- a/packages/runtime-host/src/client/registered-host-termination.ts +++ b/packages/runtime-host/src/client/registered-host-termination.ts @@ -24,6 +24,10 @@ import { } from '@maka/storage/root-authority'; import { readHostRegistration } from '../control/registration.js'; import type { HostRegistration } from '../protocol/index.js'; +import { + readRuntimeHostProcessIdentity, + type RuntimeHostProcessIdentity, +} from './process-identity.js'; const TERMINATION_SETTLE_MS = 2_000; @@ -34,22 +38,49 @@ export interface RegisteredRuntimeHostIdentity { readonly pid: number; } -interface RegisteredRuntimeHostTerminationDependencies { - readonly terminateProcess: typeof terminateProcessTree; +interface RuntimeHostExitDependencies { readonly isProcessAlive: (pid: number) => boolean; readonly settleMs: number; } +interface RegisteredRuntimeHostTerminationDependencies extends RuntimeHostExitDependencies { + readonly terminateProcess: typeof terminateProcessTree; +} + +export interface ObservedRegisteredRuntimeHostTerminationAuthority { + readonly processIdentity: RuntimeHostProcessIdentity; + readonly isCurrent: () => boolean; +} + +export interface ObservedRegisteredRuntimeHost { + readonly rootPath: string; + readonly registration: HostRegistration; +} + +interface ObservedRegisteredRuntimeHostTerminationDependencies extends RuntimeHostExitDependencies { + readonly readProcessIdentity: typeof readRuntimeHostProcessIdentity; + readonly readRegistration: typeof readHostRegistration; + readonly signalProcess: (pid: number) => boolean; +} + const defaultDependencies: RegisteredRuntimeHostTerminationDependencies = { terminateProcess: terminateProcessTree, isProcessAlive, settleMs: TERMINATION_SETTLE_MS, }; +const defaultObservedDependencies: ObservedRegisteredRuntimeHostTerminationDependencies = { + isProcessAlive, + readProcessIdentity: readRuntimeHostProcessIdentity, + readRegistration: readHostRegistration, + signalProcess: forceKillProcess, + settleMs: TERMINATION_SETTLE_MS, +}; + /** * Force-terminates only the exact local ephemeral Host still registered for * the expected State Root. Callers must reserve this for explicit recovery - * after graceful retirement fails. + * and keep their authorization current until the signal is sent. */ export function forceTerminateRegisteredRuntimeHost( identity: RegisteredRuntimeHostIdentity, @@ -103,6 +134,69 @@ export async function forceTerminateRegisteredRuntimeHostWithDependencies( return waitForExit(identity.pid, dependencies); } +/** + * Stops an ephemeral Host that Desktop did not launch in this process. Unlike + * the owned-process path above, its authority is limited to the exact root PID + * whose OS process lifetime was observed across a valid handshake. + */ +export function forceTerminateObservedRegisteredRuntimeHost( + observed: ObservedRegisteredRuntimeHost, + authority: ObservedRegisteredRuntimeHostTerminationAuthority, +): Promise { + return forceTerminateObservedRegisteredRuntimeHostWithDependencies( + observed, + authority, + defaultObservedDependencies, + ); +} + +export async function forceTerminateObservedRegisteredRuntimeHostWithDependencies( + observed: ObservedRegisteredRuntimeHost, + authority: ObservedRegisteredRuntimeHostTerminationAuthority, + dependencies: ObservedRegisteredRuntimeHostTerminationDependencies, +): Promise { + const { registration } = observed; + if ( + registration.lifecycleMode !== 'ephemeral' || + !authority.isCurrent() || + authority.processIdentity.startIdentity.length === 0 + ) { + return false; + } + const capability = await resolveStorageRoot({ path: observed.rootPath, kind: 'interactive' }); + if (capability.rootId !== registration.rootId) return false; + const { controlDirectory } = await prepareStorageRootControlDirectory(capability); + const signalTarget = await dependencies.readRegistration(controlDirectory); + if (!signalTarget) return true; + if (!matchesObservedRegistration(signalTarget, registration) || !authority.isCurrent()) { + return false; + } + if (!dependencies.isProcessAlive(registration.pid)) return true; + const processIdentity = await dependencies.readProcessIdentity(registration.pid); + if ( + processIdentity?.startIdentity !== authority.processIdentity.startIdentity || + !authority.isCurrent() + ) { + return false; + } + const signaled = dependencies.signalProcess(registration.pid); + if (!signaled && dependencies.isProcessAlive(registration.pid)) return false; + return waitForExit(registration.pid, dependencies); +} + +function matchesObservedRegistration( + current: HostRegistration, + observed: HostRegistration, +): boolean { + return ( + current.rootId === observed.rootId && + current.hostEpoch === observed.hostEpoch && + current.pid === observed.pid && + current.lifecycleMode === observed.lifecycleMode && + current.createdAt === observed.createdAt + ); +} + function matchesIdentity( registration: HostRegistration | undefined, identity: RegisteredRuntimeHostIdentity, @@ -117,7 +211,7 @@ function matchesIdentity( async function waitForExit( pid: number, - dependencies: RegisteredRuntimeHostTerminationDependencies, + dependencies: RuntimeHostExitDependencies, ): Promise { const deadline = Date.now() + dependencies.settleMs; while (dependencies.isProcessAlive(pid)) { @@ -139,3 +233,12 @@ function isProcessAlive(pid: number): boolean { ); } } + +function forceKillProcess(pid: number): boolean { + try { + process.kill(pid, 'SIGKILL'); + return true; + } catch { + return false; + } +} diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts index cd1ff4003f..7c2ac17d11 100644 --- a/packages/runtime-host/src/transport/peer-native.ts +++ b/packages/runtime-host/src/transport/peer-native.ts @@ -143,6 +143,7 @@ export interface RuntimeHostPeerTransitRelayCandidate { } interface RuntimeHostPeerNativeModule { + readProcessStartIdentity?(pid: number): unknown; ensurePeerIdentity(keyPath: string): Promise; signPeerIdentity( keyPath: string, @@ -166,6 +167,16 @@ interface RuntimeHostPeerNativeModule { }): unknown; } +export function readRuntimeHostNativeProcessStartIdentity( + nativePath: string, + pid: number, +): string | undefined { + const reader = loadNativeModule(nativePath).readProcessStartIdentity; + if (!reader) return undefined; + const identity = reader(pid); + return typeof identity === 'string' && isProcessStartIdentity(identity) ? identity : undefined; +} + export async function signRuntimeHostPeerIdentity(input: { readonly nativePath: string; readonly keyPath: string; @@ -669,6 +680,12 @@ function isPeerId(value: unknown): value is string { ); } +function isProcessStartIdentity(value: string): boolean { + return ( + Buffer.byteLength(value, 'utf8') <= 256 && /^(?:darwin|linux|windows):[0-9a-f:-]+$/u.test(value) + ); +} + function asError(error: unknown): Error { return error instanceof Error ? error : new Error(String(error)); }