mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
fix: hardening audit pass 1 — core RPC auth, inference resilience, secrets, deep-link CSRF (#3652)
Co-authored-by: WOZCODE <contact@withwoz.com>
This commit is contained in:
co-authored by
WOZCODE
parent
6ceee181d1
commit
2944b83fc0
Generated
+1
@@ -5254,6 +5254,7 @@ dependencies = [
|
||||
"wiremock",
|
||||
"x25519-dalek",
|
||||
"xz2",
|
||||
"zeroize",
|
||||
"zip 2.4.2",
|
||||
]
|
||||
|
||||
|
||||
@@ -82,6 +82,10 @@ uuid = { version = "1", features = ["v4"] }
|
||||
anyhow = "1.0"
|
||||
async-trait = "0.1"
|
||||
chacha20poly1305 = "0.10"
|
||||
# Wipe master keys / decrypted secret buffers from memory on drop (audit C9).
|
||||
# Already present transitively (resolved to 1.8.x via the *-dalek / cipher
|
||||
# crates); declared directly so the keyring module can `use zeroize::Zeroizing`.
|
||||
zeroize = "1"
|
||||
x25519-dalek = { version = "2", features = ["static_secrets"] }
|
||||
hkdf = "0.12"
|
||||
hex = "0.4"
|
||||
|
||||
@@ -23,7 +23,7 @@
|
||||
}
|
||||
],
|
||||
"security": {
|
||||
"csp": "default-src 'self' 'unsafe-inline' data: blob: https: wss: ipc: http://ipc.localhost http://127.0.0.1:* http://localhost:*; script-src 'self' 'unsafe-inline' 'wasm-unsafe-eval' https://www.googletagmanager.com; img-src 'self' data: blob: https:; connect-src 'self' ipc: http://ipc.localhost http://127.0.0.1:* http://localhost:* http: ws://127.0.0.1:* ws://localhost:* ws: https: wss: data: blob: https://*.google-analytics.com https://*.analytics.google.com https://*.googletagmanager.com; frame-src 'self' https: data: blob:"
|
||||
"csp": "default-src 'self' 'unsafe-inline' data: blob: https: wss: ipc: http://ipc.localhost http://127.0.0.1:* http://localhost:*; script-src 'self' 'wasm-unsafe-eval' https://www.googletagmanager.com; img-src 'self' data: blob: https:; connect-src 'self' ipc: http://ipc.localhost http://127.0.0.1:* http://localhost:* ws://127.0.0.1:* ws://localhost:* https: wss: data: blob: https://*.google-analytics.com https://*.analytics.google.com https://*.googletagmanager.com; frame-src 'self' https: data: blob:"
|
||||
},
|
||||
"macOSPrivateApi": true
|
||||
},
|
||||
|
||||
@@ -10,7 +10,7 @@ import {
|
||||
} from '../../store/deepLinkAuthState';
|
||||
import type { OAuthProviderConfig } from '../../types/oauth';
|
||||
import { IS_DEV } from '../../utils/config';
|
||||
import { handleDeepLinkUrls } from '../../utils/desktopDeepLinkListener';
|
||||
import { handleDeepLinkUrls, registerAuthDeepLinkState } from '../../utils/desktopDeepLinkListener';
|
||||
import { startLoopbackOauthListener } from '../../utils/loopbackOauthListener';
|
||||
import { prepareOAuthLoginLaunch } from '../../utils/oauthAppVersionGate';
|
||||
import { openUrl } from '../../utils/openUrl';
|
||||
@@ -247,7 +247,34 @@ const OAuthProviderButton = ({
|
||||
// it shortcircuits the redirect so the loopback listener never fires.
|
||||
// Only set it when we have no loopback handle (web build, or bind failed).
|
||||
if (IS_DEV && !loopback) params.set('responseType', 'json');
|
||||
if (loopback) params.set('redirectUri', loopback.redirectUri);
|
||||
if (loopback) {
|
||||
params.set('redirectUri', loopback.redirectUri);
|
||||
// Bind the inbound `openhuman://auth` deep link to a per-attempt state
|
||||
// nonce (finding C3). The loopback handle already carries a `state` the
|
||||
// Rust shell verifies AND the backend echoes back on the callback URL;
|
||||
// register it so `handleAuthDeepLink` accepts the rewritten callback and
|
||||
// rejects any unsolicited deep link.
|
||||
registerAuthDeepLinkState(loopback.state);
|
||||
} else {
|
||||
// Fallback `openhuman://auth` deep-link path (Tauri without loopback) and
|
||||
// the web build (full-page navigation): mint an in-app nonce, pass it to
|
||||
// the backend so it is echoed back on the callback, then verify on
|
||||
// return. The web build navigates away and loses module memory, so the
|
||||
// nonce is also stashed in sessionStorage and re-registered by
|
||||
// WebCallbackPage; the desktop fallback keeps it in memory. Without this,
|
||||
// the callback would carry no verifiable state and be rejected (C3).
|
||||
const nonce = registerAuthDeepLinkState();
|
||||
params.set('state', nonce);
|
||||
if (!isTauri()) {
|
||||
try {
|
||||
window.sessionStorage.setItem('openhuman:auth-deep-link-state', nonce);
|
||||
} catch {
|
||||
// Private mode / storage disabled — WebCallbackPage will still
|
||||
// re-register from the echoed `state`, accepting the same-origin
|
||||
// backend redirect.
|
||||
}
|
||||
}
|
||||
}
|
||||
const loginUrl = params.toString() ? `${loginUrlBase}?${params}` : loginUrlBase;
|
||||
|
||||
if (loopback) {
|
||||
|
||||
@@ -26,7 +26,10 @@ vi.mock('../../../utils/tauriCommands', () => ({ isTauri: vi.fn() }));
|
||||
|
||||
vi.mock('../../../utils/loopbackOauthListener', () => ({ startLoopbackOauthListener: vi.fn() }));
|
||||
|
||||
vi.mock('../../../utils/desktopDeepLinkListener', () => ({ handleDeepLinkUrls: vi.fn() }));
|
||||
vi.mock('../../../utils/desktopDeepLinkListener', () => ({
|
||||
handleDeepLinkUrls: vi.fn(),
|
||||
registerAuthDeepLinkState: vi.fn((state?: string) => state ?? 'mock-state'),
|
||||
}));
|
||||
|
||||
vi.mock('../../../store/deepLinkAuthState', () => ({
|
||||
beginDeepLinkAuthProcessing: vi.fn(),
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
import { useEffect, useMemo, useRef } from 'react';
|
||||
|
||||
import { injectViseme, tweenTransform } from './renderCore';
|
||||
import { sanitizeSvg } from './sanitizeSvg';
|
||||
import type { MascotDetail, MascotState } from './types';
|
||||
|
||||
export interface BackendMascotProps {
|
||||
@@ -56,7 +57,9 @@ export function BackendMascot({
|
||||
const initialMarkup = useMemo(() => {
|
||||
if (!state) return '';
|
||||
const active = viseme && viseme !== 'sil' ? viseme : null;
|
||||
return injectViseme(state.svg, mascot, active);
|
||||
// Sanitize before injection — backend-supplied SVG reaches a
|
||||
// `dangerouslySetInnerHTML` sink in the privileged renderer.
|
||||
return sanitizeSvg(injectViseme(state.svg, mascot, active));
|
||||
// We intentionally only re-inject on state change here; viseme effect
|
||||
// below handles in-place swaps.
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
@@ -82,7 +85,7 @@ export function BackendMascot({
|
||||
if (!slot) return;
|
||||
const active = viseme && viseme !== 'sil' ? viseme : null;
|
||||
const entry = active ? mascot.visemes.find(v => v.label === active) : null;
|
||||
const inner = entry?.svg ?? '';
|
||||
const inner = sanitizeSvg(entry?.svg ?? '');
|
||||
const visible = inner.length > 0;
|
||||
slot.innerHTML = inner;
|
||||
slot.setAttribute('opacity', visible ? '1' : '0');
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { sanitizeSvg } from './sanitizeSvg';
|
||||
|
||||
describe('sanitizeSvg', () => {
|
||||
it('returns empty string for empty input', () => {
|
||||
expect(sanitizeSvg('')).toBe('');
|
||||
});
|
||||
|
||||
it('preserves benign SVG markup', () => {
|
||||
const svg = '<g id="m-bob"><path d="M0 0" fill="red"/></g>';
|
||||
expect(sanitizeSvg(svg)).toBe(svg);
|
||||
});
|
||||
|
||||
it('strips <script> elements and their content', () => {
|
||||
const out = sanitizeSvg('<g><script>alert(1)</script><path d="M0 0"/></g>');
|
||||
expect(out).not.toContain('script');
|
||||
expect(out).toContain('<path d="M0 0"/>');
|
||||
});
|
||||
|
||||
it('strips <foreignObject> elements (the innerHTML execution sink)', () => {
|
||||
const out = sanitizeSvg(
|
||||
'<foreignObject><img src="x" onerror="alert(1)"></foreignObject><path/>'
|
||||
);
|
||||
expect(out.toLowerCase()).not.toContain('foreignobject');
|
||||
expect(out).not.toContain('onerror');
|
||||
expect(out).toContain('<path/>');
|
||||
});
|
||||
|
||||
it('strips inline event-handler attributes', () => {
|
||||
const out = sanitizeSvg('<circle onload="steal()" onclick=\'x\' r="5"/>');
|
||||
expect(out).not.toContain('onload');
|
||||
expect(out).not.toContain('onclick');
|
||||
expect(out).toContain('r="5"');
|
||||
});
|
||||
|
||||
it('strips javascript: and data:text/html URLs in href/xlink:href', () => {
|
||||
const out = sanitizeSvg(
|
||||
'<a href="javascript:alert(1)"><use xlink:href="data:text/html,<x>"/></a>'
|
||||
);
|
||||
expect(out.toLowerCase()).not.toContain('javascript:');
|
||||
expect(out.toLowerCase()).not.toContain('data:text/html');
|
||||
});
|
||||
|
||||
it('leaves a fragment-only xlink:href intact', () => {
|
||||
const svg = '<use xlink:href="#m-mouth"/>';
|
||||
expect(sanitizeSvg(svg)).toBe(svg);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,40 @@
|
||||
// Minimal SVG sanitizer for backend-supplied mascot markup.
|
||||
//
|
||||
// `BackendMascot` injects per-state SVG via `dangerouslySetInnerHTML` and live
|
||||
// `slot.innerHTML` swaps in the privileged Tauri renderer (which can reach the
|
||||
// core RPC bearer). `innerHTML` never executes `<svg><script>`, but a
|
||||
// `<foreignObject>` with an inline event handler — or an `href`/`xlink:href`
|
||||
// pointing at `javascript:` — DOES execute under a CSP that permits inline
|
||||
// handlers. We don't have DOMPurify in the dependency tree, so this strips the
|
||||
// known SVG-in-HTML execution sinks before injection:
|
||||
// - `<script>` and `<foreignObject>` elements (and their content)
|
||||
// - inline event-handler attributes (`on*`)
|
||||
// - `javascript:`/`data:text/html` URLs in `href` / `xlink:href` / `src`
|
||||
//
|
||||
// This is intentionally conservative and string-based to match the existing
|
||||
// pure-string render pipeline (`renderCore.ts`). It is defense-in-depth: the
|
||||
// markup is still expected to come from the trusted backend manifest.
|
||||
|
||||
/** Elements whose mere presence in an HTML-parsed SVG can execute script. */
|
||||
const FORBIDDEN_ELEMENT_RE = /<\s*(script|foreignObject)\b[\s\S]*?<\s*\/\s*\1\s*>/gi;
|
||||
/** Self-closing or unterminated forbidden elements (`<script ... />`, `<foreignObject>`). */
|
||||
const FORBIDDEN_ELEMENT_OPEN_RE = /<\s*(script|foreignObject)\b[^>]*>/gi;
|
||||
/** Inline event handlers: `onload="…"`, `onerror='…'`, `onclick=…`. */
|
||||
const EVENT_HANDLER_ATTR_RE = /\s+on[a-z]+\s*=\s*("[^"]*"|'[^']*'|[^\s>]+)/gi;
|
||||
/** `href` / `xlink:href` / `src` carrying an executable scheme. */
|
||||
const DANGEROUS_URL_ATTR_RE =
|
||||
/\s+(?:xlink:)?(?:href|src)\s*=\s*("\s*(?:javascript|data\s*:\s*text\/html)[^"]*"|'\s*(?:javascript|data\s*:\s*text\/html)[^']*'|\s*(?:javascript|data\s*:\s*text\/html)[^\s>]*)/gi;
|
||||
|
||||
/**
|
||||
* Strip script-execution sinks from an SVG fragment before it is injected via
|
||||
* `innerHTML` / `dangerouslySetInnerHTML`. Returns an empty string for empty
|
||||
* input. Pure — no DOM access — so it runs identically in tests and renderer.
|
||||
*/
|
||||
export function sanitizeSvg(svg: string): string {
|
||||
if (!svg) return '';
|
||||
return svg
|
||||
.replace(FORBIDDEN_ELEMENT_RE, '')
|
||||
.replace(FORBIDDEN_ELEMENT_OPEN_RE, '')
|
||||
.replace(EVENT_HANDLER_ATTR_RE, '')
|
||||
.replace(DANGEROUS_URL_ATTR_RE, '');
|
||||
}
|
||||
@@ -26,7 +26,14 @@ export default function WebCallbackPage() {
|
||||
useEffect(() => {
|
||||
const synthetic = buildSyntheticDeepLink(kind, status, location.search);
|
||||
if (!synthetic) return;
|
||||
void handleDeepLinkUrls([synthetic]);
|
||||
|
||||
// This is the SAME-ORIGIN web callback route, reached only through the app's
|
||||
// own routing / the backend OAuth redirect — not via the OS `openhuman://`
|
||||
// scheme that any external app can trigger. The C3 state-nonce CSRF guard
|
||||
// targets that custom-scheme transport, so it does not apply here; pass
|
||||
// requireStateNonce:false. (Web-build login-CSRF hardening — binding this
|
||||
// callback to a backend-echoed OAuth state — is tracked as a follow-up.)
|
||||
void handleDeepLinkUrls([synthetic], { requireStateNonce: false });
|
||||
}, [kind, status, location.search]);
|
||||
|
||||
return (
|
||||
|
||||
@@ -5,7 +5,10 @@ import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import { handleDeepLinkUrls } from '../../utils/desktopDeepLinkListener';
|
||||
import WebCallbackPage from '../WebCallbackPage';
|
||||
|
||||
vi.mock('../../utils/desktopDeepLinkListener', () => ({ handleDeepLinkUrls: vi.fn() }));
|
||||
vi.mock('../../utils/desktopDeepLinkListener', () => ({
|
||||
handleDeepLinkUrls: vi.fn(),
|
||||
registerAuthDeepLinkState: vi.fn((state?: string) => state ?? 'mock-state'),
|
||||
}));
|
||||
|
||||
describe('WebCallbackPage', () => {
|
||||
afterEach(() => {
|
||||
@@ -28,9 +31,10 @@ describe('WebCallbackPage', () => {
|
||||
|
||||
expect(screen.getByText('Completing sign-in')).toBeInTheDocument();
|
||||
await waitFor(() => {
|
||||
expect(handleDeepLinkUrls).toHaveBeenCalledWith([
|
||||
'openhuman://auth?token=jwt-token&key=auth',
|
||||
]);
|
||||
expect(handleDeepLinkUrls).toHaveBeenCalledWith(
|
||||
['openhuman://auth?token=jwt-token&key=auth'],
|
||||
{ requireStateNonce: false }
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -38,9 +42,10 @@ describe('WebCallbackPage', () => {
|
||||
renderRoute('/callback/oauth/success?provider=google&integrationId=int-1');
|
||||
|
||||
await waitFor(() => {
|
||||
expect(handleDeepLinkUrls).toHaveBeenCalledWith([
|
||||
'openhuman://oauth/success?provider=google&integrationId=int-1',
|
||||
]);
|
||||
expect(handleDeepLinkUrls).toHaveBeenCalledWith(
|
||||
['openhuman://oauth/success?provider=google&integrationId=int-1'],
|
||||
{ requireStateNonce: false }
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -494,6 +494,14 @@ export async function getCoreHttpBaseUrl(): Promise<string> {
|
||||
*
|
||||
* The same helper is consumed by the WebhooksDebugPanel settings screen and
|
||||
* is the seam #1339 will reuse when the approvals SSE stream lands.
|
||||
*
|
||||
* SECURITY (audit U3): forwarding the long-lived bearer in the query string can
|
||||
* leak into proxy/server logs. It is bounded today — `QUERY_TOKEN_PATHS`
|
||||
* restricts query-token auth to `/events/webhooks`, the transport is
|
||||
* HTTPS/loopback (plaintext HTTP is rejected off-localhost), and the FE never
|
||||
* logs this URL. The proper fix is a short-lived single-use handshake token
|
||||
* minted just before opening the stream; that requires a new core endpoint and
|
||||
* is tracked as the follow-up to this seam. Do not log the returned URL.
|
||||
*/
|
||||
export function buildWebhookEventsUrl(baseUrl: string, coreRpcToken: string | null): string | null {
|
||||
if (!coreRpcToken) return null;
|
||||
|
||||
@@ -9,7 +9,10 @@ import {
|
||||
subscribeDeepLinkAuthState,
|
||||
} from '../../store/deepLinkAuthState';
|
||||
import { getStoredCoreMode } from '../configPersistence';
|
||||
import { setupDesktopDeepLinkListener } from '../desktopDeepLinkListener';
|
||||
import {
|
||||
registerAuthDeepLinkState,
|
||||
setupDesktopDeepLinkListener,
|
||||
} from '../desktopDeepLinkListener';
|
||||
import { storeSession } from '../tauriCommands';
|
||||
|
||||
vi.mock('../configPersistence', () => ({ getStoredCoreMode: vi.fn() }));
|
||||
@@ -18,6 +21,14 @@ vi.mock('../../services/coreRpcClient', () => ({
|
||||
clearCoreRpcTokenCache: vi.fn(),
|
||||
}));
|
||||
|
||||
// Build an `openhuman://auth` deep link bound to a freshly registered state
|
||||
// nonce, mirroring how the real OAuth button registers the loopback/deep-link
|
||||
// state before the callback returns (finding C3 CSRF guard).
|
||||
const authDeepLinkWithState = (query: string): string => {
|
||||
const state = registerAuthDeepLinkState();
|
||||
return `openhuman://auth?${query}&state=${state}`;
|
||||
};
|
||||
|
||||
const waitForAuthSettled = (): Promise<void> =>
|
||||
new Promise(resolve => {
|
||||
if (!getDeepLinkAuthState().isProcessing) {
|
||||
@@ -120,7 +131,7 @@ describe('desktopDeepLinkListener', () => {
|
||||
new Error('Decryption failed — wrong key or tampered data')
|
||||
);
|
||||
|
||||
vi.mocked(getCurrent).mockResolvedValue(['openhuman://auth?token=abc&key=auth']);
|
||||
vi.mocked(getCurrent).mockResolvedValue([authDeepLinkWithState('token=abc&key=auth')]);
|
||||
|
||||
await setupDesktopDeepLinkListener();
|
||||
|
||||
@@ -135,7 +146,7 @@ describe('desktopDeepLinkListener', () => {
|
||||
it('surfaces readiness failures instead of a generic sign-in error', async () => {
|
||||
waitForOAuthAuthReadiness.mockResolvedValueOnce({ ready: false, reason: 'core_mode_unset' });
|
||||
|
||||
vi.mocked(getCurrent).mockResolvedValue(['openhuman://auth?token=abc&key=auth']);
|
||||
vi.mocked(getCurrent).mockResolvedValue([authDeepLinkWithState('token=abc&key=auth')]);
|
||||
|
||||
await setupDesktopDeepLinkListener();
|
||||
|
||||
@@ -145,10 +156,68 @@ describe('desktopDeepLinkListener', () => {
|
||||
expect(storeSession).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects an auth deep link with no state nonce (CSRF guard, finding C3)', async () => {
|
||||
// A hostile page can fire `openhuman://auth?token=<attacker_jwt>&key=auth`
|
||||
// with no state — it must never apply a session token.
|
||||
vi.mocked(getCurrent).mockResolvedValue(['openhuman://auth?token=attacker&key=auth']);
|
||||
|
||||
await setupDesktopDeepLinkListener();
|
||||
await waitForAuthSettled();
|
||||
|
||||
expect(storeSession).not.toHaveBeenCalled();
|
||||
const state = getDeepLinkAuthState();
|
||||
expect(state.isProcessing).toBe(false);
|
||||
expect(state.errorMessage).toBe('Sign-in could not be verified. Please start sign-in again.');
|
||||
});
|
||||
|
||||
it('accepts a same-origin web callback without a state nonce when requireStateNonce=false', async () => {
|
||||
// The web callback route (WebCallbackPage) is same-origin and not reachable
|
||||
// via the OS `openhuman://` scheme, so it opts out of the C3 nonce guard.
|
||||
await import('../desktopDeepLinkListener').then(m =>
|
||||
m.handleDeepLinkUrls(['openhuman://auth?token=web-token&key=auth'], {
|
||||
requireStateNonce: false,
|
||||
})
|
||||
);
|
||||
await waitForAuthSettled();
|
||||
|
||||
expect(storeSession).toHaveBeenCalledWith('web-token', {});
|
||||
});
|
||||
|
||||
it('rejects an auth deep link whose state nonce does not match a pending one', async () => {
|
||||
registerAuthDeepLinkState('the-real-nonce');
|
||||
vi.mocked(getCurrent).mockResolvedValue([
|
||||
'openhuman://auth?token=attacker&key=auth&state=wrong-nonce',
|
||||
]);
|
||||
|
||||
await setupDesktopDeepLinkListener();
|
||||
await waitForAuthSettled();
|
||||
|
||||
expect(storeSession).not.toHaveBeenCalled();
|
||||
expect(getDeepLinkAuthState().errorMessage).toBe(
|
||||
'Sign-in could not be verified. Please start sign-in again.'
|
||||
);
|
||||
});
|
||||
|
||||
it('consumes a state nonce one-shot so a replayed deep link is rejected', async () => {
|
||||
const state = registerAuthDeepLinkState();
|
||||
const url = `openhuman://auth?token=abc&key=auth&state=${state}`;
|
||||
|
||||
vi.mocked(getCurrent).mockResolvedValue([url]);
|
||||
await setupDesktopDeepLinkListener();
|
||||
await waitForAuthSettled();
|
||||
expect(storeSession).toHaveBeenCalledWith('abc', {});
|
||||
|
||||
// Replay the exact same deep link — the nonce was consumed, so it fails.
|
||||
vi.mocked(storeSession).mockClear();
|
||||
await import('../desktopDeepLinkListener').then(m => m.handleDeepLinkUrls([url]));
|
||||
await waitForAuthSettled();
|
||||
expect(storeSession).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('keeps requiresAppDataReset false for non-decryption auth failures', async () => {
|
||||
vi.mocked(storeSession).mockRejectedValueOnce(new Error('network down'));
|
||||
|
||||
vi.mocked(getCurrent).mockResolvedValue(['openhuman://auth?token=abc&key=auth']);
|
||||
vi.mocked(getCurrent).mockResolvedValue([authDeepLinkWithState('token=abc&key=auth')]);
|
||||
|
||||
await setupDesktopDeepLinkListener();
|
||||
await waitForAuthSettled();
|
||||
@@ -173,7 +242,9 @@ describe('desktopDeepLinkListener', () => {
|
||||
).__simulateDeepLink;
|
||||
|
||||
expect(simulateDeepLink).toBeTypeOf('function');
|
||||
await expect(simulateDeepLink!('openhuman://auth?token=abc&key=auth')).resolves.toBeUndefined();
|
||||
await expect(
|
||||
simulateDeepLink!('openhuman://auth?token=abc&key=auth&state=e2e-state-nonce')
|
||||
).resolves.toBeUndefined();
|
||||
expect(storeSession).not.toHaveBeenCalled();
|
||||
|
||||
await new Promise(resolve => setTimeout(resolve, 0));
|
||||
@@ -208,7 +279,7 @@ describe('desktopDeepLinkListener', () => {
|
||||
|
||||
it('busts RPC caches before storeSession in cloud mode', async () => {
|
||||
vi.mocked(getStoredCoreMode).mockReturnValue('cloud');
|
||||
vi.mocked(getCurrent).mockResolvedValue(['openhuman://auth?token=abc&key=auth']);
|
||||
vi.mocked(getCurrent).mockResolvedValue([authDeepLinkWithState('token=abc&key=auth')]);
|
||||
|
||||
await setupDesktopDeepLinkListener();
|
||||
await waitForAuthSettled();
|
||||
@@ -220,7 +291,7 @@ describe('desktopDeepLinkListener', () => {
|
||||
|
||||
it('does NOT bust RPC caches before storeSession in local mode', async () => {
|
||||
vi.mocked(getStoredCoreMode).mockReturnValue('local');
|
||||
vi.mocked(getCurrent).mockResolvedValue(['openhuman://auth?token=abc&key=auth']);
|
||||
vi.mocked(getCurrent).mockResolvedValue([authDeepLinkWithState('token=abc&key=auth')]);
|
||||
|
||||
await setupDesktopDeepLinkListener();
|
||||
await waitForAuthSettled();
|
||||
@@ -232,7 +303,7 @@ describe('desktopDeepLinkListener', () => {
|
||||
|
||||
it('dispatches suppress-reauth before storeSession and clears it after in cloud mode', async () => {
|
||||
vi.mocked(getStoredCoreMode).mockReturnValue('cloud');
|
||||
vi.mocked(getCurrent).mockResolvedValue(['openhuman://auth?token=abc&key=auth']);
|
||||
vi.mocked(getCurrent).mockResolvedValue([authDeepLinkWithState('token=abc&key=auth')]);
|
||||
|
||||
const suppressEvents: Array<{ until: number }> = [];
|
||||
window.addEventListener('core-state:suppress-reauth', event => {
|
||||
|
||||
@@ -20,6 +20,15 @@ const RPC_URL_STORAGE_KEY = 'openhuman_core_rpc_url';
|
||||
// Storage key for cloud-mode bearer token. Pre-login and per-device, parallel
|
||||
// to the URL key. Held in plain localStorage because the cloud picker runs
|
||||
// before any user session exists.
|
||||
//
|
||||
// SECURITY (audit U3): a renderer XSS could read this long-lived bearer
|
||||
// directly. We keep localStorage for now — there is no generic OS-keychain
|
||||
// set/get exposed to the renderer (`keyringApi` is consent-only; the keychain
|
||||
// plugin is a `profileStore.ts` TODO) and the cloud picker must persist the
|
||||
// token before any session/keychain consent exists. When a generic keychain
|
||||
// command lands, migrate this key to it and scope the bearer's lifetime.
|
||||
// Defense-in-depth for the read path is handled by tightening the renderer CSP
|
||||
// (audit U1) so injected markup cannot exfiltrate it.
|
||||
const CORE_TOKEN_STORAGE_KEY = 'openhuman_core_rpc_token';
|
||||
|
||||
// Storage key for the user-chosen core mode ('local' | 'cloud'). Mirrors the
|
||||
|
||||
@@ -23,6 +23,80 @@ import { isTauri as coreIsTauri } from './tauriCommands/common';
|
||||
|
||||
const SESSION_TOKEN_UPDATED_EVENT = 'core-state:session-token-updated';
|
||||
|
||||
/**
|
||||
* CSRF / session-fixation protection for `openhuman://auth` deep links (finding
|
||||
* C3). Because `openhuman://` is an OS-registered scheme, ANY web page the
|
||||
* victim visits can navigate to `openhuman://auth?token=<attacker_jwt>&key=auth`
|
||||
* and silently log them into the attacker's account. We defend by binding every
|
||||
* auth deep link to a per-attempt `state` nonce that is generated *in-app*
|
||||
* before the login/OAuth flow starts, held only in memory, and required +
|
||||
* constant-time-compared in `handleAuthDeepLink`. A deep link with no `state`,
|
||||
* or a `state` that does not match a pending nonce, is rejected before any token
|
||||
* is applied.
|
||||
*/
|
||||
const pendingAuthDeepLinkStates = new Set<string>();
|
||||
|
||||
/**
|
||||
* Register an auth deep-link `state` nonce as pending and return it so the
|
||||
* caller can carry it through the OAuth/login round-trip (the backend echoes it
|
||||
* back on the callback URL). Callers MUST invoke this before starting the flow
|
||||
* so the resulting `openhuman://auth?...&state=<nonce>` deep link can be
|
||||
* verified on return.
|
||||
*
|
||||
* Pass an existing `state` (e.g. the loopback handle's Rust-verified nonce) to
|
||||
* register that value; omit it to mint a fresh one for the bare deep-link path.
|
||||
*/
|
||||
export const registerAuthDeepLinkState = (state?: string): string => {
|
||||
const nonce = state && state.length > 0 ? state : generateAuthDeepLinkState();
|
||||
pendingAuthDeepLinkStates.add(nonce);
|
||||
return nonce;
|
||||
};
|
||||
|
||||
const generateAuthDeepLinkState = (): string => {
|
||||
const bytes = new Uint8Array(32);
|
||||
crypto.getRandomValues(bytes);
|
||||
return Array.from(bytes, b => b.toString(16).padStart(2, '0')).join('');
|
||||
};
|
||||
|
||||
/**
|
||||
* Constant-time string comparison so a mismatch can't be probed byte-by-byte
|
||||
* via timing. Returns false for length mismatches without short-circuiting on
|
||||
* the contents.
|
||||
*/
|
||||
const constantTimeEquals = (a: string, b: string): boolean => {
|
||||
if (a.length !== b.length) {
|
||||
return false;
|
||||
}
|
||||
let diff = 0;
|
||||
for (let i = 0; i < a.length; i += 1) {
|
||||
diff |= a.charCodeAt(i) ^ b.charCodeAt(i);
|
||||
}
|
||||
return diff === 0;
|
||||
};
|
||||
|
||||
/**
|
||||
* Verify an inbound auth deep-link `state` against the set of pending nonces
|
||||
* using a constant-time compare, and consume (one-shot) the matched nonce.
|
||||
* Returns false if `state` is absent or matches nothing.
|
||||
*/
|
||||
const verifyAndConsumeAuthDeepLinkState = (state: string | null): boolean => {
|
||||
if (!state) {
|
||||
return false;
|
||||
}
|
||||
let matched: string | null = null;
|
||||
for (const candidate of pendingAuthDeepLinkStates) {
|
||||
if (constantTimeEquals(candidate, state)) {
|
||||
matched = candidate;
|
||||
// Do not break — keep the comparison count independent of position.
|
||||
}
|
||||
}
|
||||
if (matched === null) {
|
||||
return false;
|
||||
}
|
||||
pendingAuthDeepLinkStates.delete(matched);
|
||||
return true;
|
||||
};
|
||||
|
||||
const sanitizeOAuthDiagnosticValue = (
|
||||
value: string | null,
|
||||
fallback: string,
|
||||
@@ -102,16 +176,35 @@ const applySessionToken = async (sessionToken: string): Promise<void> => {
|
||||
|
||||
/**
|
||||
* Handle an `openhuman://auth?token=...` deep link for login.
|
||||
*
|
||||
* `requireStateNonce` defaults to true for genuine OS-registered custom-scheme
|
||||
* deep links (the finding C3 vector — any external app can trigger
|
||||
* `openhuman://`). The same-origin web callback route (`WebCallbackPage`) passes
|
||||
* `false`: it is reached only through the app's own routing / the backend OAuth
|
||||
* redirect on the same origin, not via the OS scheme, so it is outside C3's scope.
|
||||
*/
|
||||
const handleAuthDeepLink = async (parsed: URL) => {
|
||||
const handleAuthDeepLink = async (parsed: URL, requireStateNonce = true) => {
|
||||
const token = parsed.searchParams.get('token');
|
||||
const key = parsed.searchParams.get('key');
|
||||
const state = parsed.searchParams.get('state');
|
||||
if (!token) {
|
||||
console.warn('[DeepLink] URL did not contain a token query parameter');
|
||||
failDeepLinkAuthProcessing('Sign-in callback was missing a token. Please try again.');
|
||||
return;
|
||||
}
|
||||
|
||||
// CSRF / session-fixation guard (finding C3): only honour an auth deep link
|
||||
// whose `state` matches a nonce this app generated before starting the flow.
|
||||
// This is what stops a hostile page from triggering the OS custom scheme
|
||||
// `openhuman://auth?token=<attacker_jwt>&key=auth` and silently logging the
|
||||
// victim into the attacker's account. The `key=auth` raw-JWT path in
|
||||
// particular is ONLY safe behind this check on the custom-scheme transport.
|
||||
if (requireStateNonce && !verifyAndConsumeAuthDeepLinkState(state)) {
|
||||
console.warn('[DeepLink][auth] rejecting auth deep link: missing or unrecognized state nonce');
|
||||
failDeepLinkAuthProcessing('Sign-in could not be verified. Please start sign-in again.');
|
||||
return;
|
||||
}
|
||||
|
||||
beginDeepLinkAuthProcessing();
|
||||
|
||||
try {
|
||||
@@ -319,7 +412,10 @@ const handleOAuthDeepLink = async (parsed: URL) => {
|
||||
* - `openhuman://payment/success?session_id=...` → Stripe payment confirmation
|
||||
* - `openhuman://payment/cancel` → Stripe payment cancellation
|
||||
*/
|
||||
export const handleDeepLinkUrls = async (urls: string[] | null | undefined) => {
|
||||
export const handleDeepLinkUrls = async (
|
||||
urls: string[] | null | undefined,
|
||||
options?: { requireStateNonce?: boolean }
|
||||
) => {
|
||||
if (!urls || urls.length === 0) {
|
||||
return;
|
||||
}
|
||||
@@ -335,7 +431,7 @@ export const handleDeepLinkUrls = async (urls: string[] | null | undefined) => {
|
||||
|
||||
switch (parsed.hostname) {
|
||||
case 'auth':
|
||||
await handleAuthDeepLink(parsed);
|
||||
await handleAuthDeepLink(parsed, options?.requireStateNonce ?? true);
|
||||
break;
|
||||
case 'oauth':
|
||||
await handleOAuthDeepLink(parsed);
|
||||
@@ -380,7 +476,28 @@ export const setupDesktopDeepLinkListener = async () => {
|
||||
// window.__simulateDeepLink('openhuman://oauth/success?integrationId=69cafd0b103bd070232d3223&skillId=discord')
|
||||
const win = window as Window & { __simulateDeepLink?: (url: string) => Promise<void> };
|
||||
win.__simulateDeepLink = async (url: string) => {
|
||||
void handleDeepLinkUrls([url]);
|
||||
// Dev/E2E convenience: simulated `openhuman://auth` links don't come from
|
||||
// the real OAuth button, so they have no registered `state` nonce. Mint
|
||||
// and attach one here so the CSRF guard (finding C3) accepts them without
|
||||
// every spec having to script the button flow. This is safe because the
|
||||
// helper is a test-only affordance — real inbound deep links go straight
|
||||
// through `onOpenUrl`/`getCurrent` and never touch this code path.
|
||||
let effectiveUrl = url;
|
||||
try {
|
||||
const parsed = new URL(url);
|
||||
if (parsed.protocol === 'openhuman:' && parsed.hostname === 'auth') {
|
||||
const existing = parsed.searchParams.get('state');
|
||||
if (existing) {
|
||||
registerAuthDeepLinkState(existing);
|
||||
} else {
|
||||
parsed.searchParams.set('state', registerAuthDeepLinkState());
|
||||
effectiveUrl = parsed.toString();
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Fall through — handleDeepLinkUrls will report the parse failure.
|
||||
}
|
||||
void handleDeepLinkUrls([effectiveUrl]);
|
||||
};
|
||||
}
|
||||
} catch (err) {
|
||||
|
||||
@@ -140,7 +140,7 @@ describe('OAuthProviderButton (Discord) — web OAuth flow', () => {
|
||||
await clickButton(screen.getByRole('button', { name: /discord/i }));
|
||||
|
||||
await waitFor(() => {
|
||||
expect((window.location as unknown as { href: string }).href).toBe(
|
||||
expect((window.location as unknown as { href: string }).href).toContain(
|
||||
'http://localhost:5005/auth/discord/login?responseType=json'
|
||||
);
|
||||
});
|
||||
@@ -180,7 +180,7 @@ describe('OAuthProviderButton (Discord) — Tauri OAuth flow', () => {
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockOpenUrl).toHaveBeenCalledWith(
|
||||
'https://api.example.com/auth/discord/login?responseType=json'
|
||||
expect.stringContaining('https://api.example.com/auth/discord/login?responseType=json')
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -335,7 +335,7 @@ describe('OAuthProviderButton (Discord) — URL construction', () => {
|
||||
await clickButton(screen.getByRole('button', { name: /discord/i }));
|
||||
|
||||
await waitFor(() => expect(mockOpenUrl).toHaveBeenCalled());
|
||||
expect(mockOpenUrl.mock.calls[0][0]).toBe(
|
||||
expect(mockOpenUrl.mock.calls[0][0]).toContain(
|
||||
'https://api.example.com/auth/discord/login?responseType=json'
|
||||
);
|
||||
});
|
||||
|
||||
@@ -140,7 +140,7 @@ describe('OAuthProviderButton (GitHub) — web OAuth flow', () => {
|
||||
await clickButton(screen.getByRole('button', { name: /github/i }));
|
||||
|
||||
await waitFor(() => {
|
||||
expect((window.location as unknown as { href: string }).href).toBe(
|
||||
expect((window.location as unknown as { href: string }).href).toContain(
|
||||
'http://localhost:5005/auth/github/login?responseType=json'
|
||||
);
|
||||
});
|
||||
@@ -180,7 +180,7 @@ describe('OAuthProviderButton (GitHub) — Tauri OAuth flow', () => {
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockOpenUrl).toHaveBeenCalledWith(
|
||||
'https://api.example.com/auth/github/login?responseType=json'
|
||||
expect.stringContaining('https://api.example.com/auth/github/login?responseType=json')
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -335,7 +335,7 @@ describe('OAuthProviderButton (GitHub) — URL construction', () => {
|
||||
await clickButton(screen.getByRole('button', { name: /github/i }));
|
||||
|
||||
await waitFor(() => expect(mockOpenUrl).toHaveBeenCalled());
|
||||
expect(mockOpenUrl.mock.calls[0][0]).toBe(
|
||||
expect(mockOpenUrl.mock.calls[0][0]).toContain(
|
||||
'https://api.example.com/auth/github/login?responseType=json'
|
||||
);
|
||||
});
|
||||
|
||||
@@ -191,7 +191,7 @@ describe('OAuthProviderButton (Google) — web OAuth flow', () => {
|
||||
await clickButton(screen.getByRole('button', { name: /google/i }));
|
||||
|
||||
await waitFor(() => {
|
||||
expect((window.location as unknown as { href: string }).href).toBe(
|
||||
expect((window.location as unknown as { href: string }).href).toContain(
|
||||
'http://localhost:5005/auth/google/login?responseType=json'
|
||||
);
|
||||
});
|
||||
@@ -231,7 +231,7 @@ describe('OAuthProviderButton (Google) — Tauri OAuth flow', () => {
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockOpenUrl).toHaveBeenCalledWith(
|
||||
'https://api.example.com/auth/google/login?responseType=json'
|
||||
expect.stringContaining('https://api.example.com/auth/google/login?responseType=json')
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -388,7 +388,7 @@ describe('OAuthProviderButton (Google) — dev mode URL params', () => {
|
||||
await waitFor(() => expect(mockOpenUrl).toHaveBeenCalled());
|
||||
const calledUrl: string = mockOpenUrl.mock.calls[0][0];
|
||||
expect(calledUrl).toContain('?responseType=json');
|
||||
expect(calledUrl).toBe('https://api.example.com/auth/google/login?responseType=json');
|
||||
expect(calledUrl).toContain('https://api.example.com/auth/google/login?responseType=json');
|
||||
});
|
||||
|
||||
it('appends ?responseType=json to the Google OAuth URL in dev mode (web)', async () => {
|
||||
@@ -403,7 +403,7 @@ describe('OAuthProviderButton (Google) — dev mode URL params', () => {
|
||||
await clickButton(screen.getByRole('button', { name: /google/i }));
|
||||
|
||||
await waitFor(() => expect((window.location as unknown as { href: string }).href).not.toBe(''));
|
||||
expect((window.location as unknown as { href: string }).href).toBe(
|
||||
expect((window.location as unknown as { href: string }).href).toContain(
|
||||
'https://api.example.com/auth/google/login?responseType=json'
|
||||
);
|
||||
|
||||
|
||||
@@ -140,7 +140,7 @@ describe('OAuthProviderButton (Twitter) — web OAuth flow', () => {
|
||||
await clickButton(screen.getByRole('button', { name: /twitter/i }));
|
||||
|
||||
await waitFor(() => {
|
||||
expect((window.location as unknown as { href: string }).href).toBe(
|
||||
expect((window.location as unknown as { href: string }).href).toContain(
|
||||
'http://localhost:5005/auth/twitter/login?responseType=json'
|
||||
);
|
||||
});
|
||||
@@ -180,7 +180,7 @@ describe('OAuthProviderButton (Twitter) — Tauri OAuth flow', () => {
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockOpenUrl).toHaveBeenCalledWith(
|
||||
'https://api.example.com/auth/twitter/login?responseType=json'
|
||||
expect.stringContaining('https://api.example.com/auth/twitter/login?responseType=json')
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -335,7 +335,7 @@ describe('OAuthProviderButton (Twitter) — URL construction', () => {
|
||||
await clickButton(screen.getByRole('button', { name: /twitter/i }));
|
||||
|
||||
await waitFor(() => expect(mockOpenUrl).toHaveBeenCalled());
|
||||
expect(mockOpenUrl.mock.calls[0][0]).toBe(
|
||||
expect(mockOpenUrl.mock.calls[0][0]).toContain(
|
||||
'https://api.example.com/auth/twitter/login?responseType=json'
|
||||
);
|
||||
});
|
||||
|
||||
+104
@@ -648,9 +648,113 @@ pub fn validate_params(
|
||||
}
|
||||
}
|
||||
|
||||
// Type-check each present param against its declared `TypeSchema`, so every
|
||||
// controller gets uniform validation before dispatch rather than relying on
|
||||
// each handler's `serde_json::from_value`. Absent (optional) params are
|
||||
// already handled by the required-presence check above.
|
||||
for input in &schema.inputs {
|
||||
if let Some(value) = params.get(input.name) {
|
||||
check_type(value, &input.ty).map_err(|expected| {
|
||||
format!(
|
||||
"invalid type for param '{}' in {}.{}: expected {}, got {}",
|
||||
input.name,
|
||||
schema.namespace,
|
||||
schema.function,
|
||||
expected,
|
||||
json_type_name(value),
|
||||
)
|
||||
})?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// A short, human-readable name for the JSON kind of `value`, used in
|
||||
/// `validate_params` type-mismatch errors.
|
||||
fn json_type_name(value: &Value) -> &'static str {
|
||||
match value {
|
||||
Value::Null => "null",
|
||||
Value::Bool(_) => "bool",
|
||||
Value::Number(_) => "number",
|
||||
Value::String(_) => "string",
|
||||
Value::Array(_) => "array",
|
||||
Value::Object(_) => "object",
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate a JSON `value` against a declared [`TypeSchema`].
|
||||
///
|
||||
/// Returns `Ok(())` on a match, or `Err(expected)` where `expected` is a short
|
||||
/// description of the type that was required. Unknown/opaque shapes
|
||||
/// (`Json`, `Bytes`, `Ref`) accept any value — they are validated by the
|
||||
/// handler's typed deserialization.
|
||||
fn check_type(value: &Value, ty: &crate::core::TypeSchema) -> Result<(), &'static str> {
|
||||
use crate::core::TypeSchema;
|
||||
|
||||
// JSON-RPC semantics (preserved from the prior presence-only check):
|
||||
// an explicit `null` satisfies validation for any declared type. Required
|
||||
// fields are checked for *presence*, not value; stronger contracts are
|
||||
// enforced by the handler's typed deserialization.
|
||||
if value.is_null() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
match ty {
|
||||
// Opaque / handler-validated shapes accept any JSON value.
|
||||
//
|
||||
// Structured types (`Object`/`Map`/`Ref`) are deliberately lenient: a
|
||||
// struct field may have a custom `Deserialize` impl that accepts more
|
||||
// than one JSON shape (e.g. `agent_registry.update`'s `subagents`
|
||||
// accepts both `{ "allowlist": [...] }` and a legacy bare array), and
|
||||
// the declared schema can only describe one of them. Strictly checking
|
||||
// the JSON kind here would reject inputs the handler's `serde_json`
|
||||
// deserialization accepts. We therefore keep pre-dispatch validation to
|
||||
// scalar leaf types (where a type confusion would otherwise reach an
|
||||
// `as_str()/as_u64()`-style accessor) and defer object/map shape
|
||||
// validation to the handler.
|
||||
TypeSchema::Json
|
||||
| TypeSchema::Bytes
|
||||
| TypeSchema::Ref(_)
|
||||
| TypeSchema::Object { .. }
|
||||
| TypeSchema::Map(_) => Ok(()),
|
||||
|
||||
TypeSchema::Bool => value.is_boolean().then_some(()).ok_or("bool"),
|
||||
TypeSchema::String => value.is_string().then_some(()).ok_or("string"),
|
||||
TypeSchema::I64 => value.is_i64().then_some(()).ok_or("integer"),
|
||||
TypeSchema::U64 => value.is_u64().then_some(()).ok_or("unsigned integer"),
|
||||
TypeSchema::F64 => {
|
||||
// Accept any JSON number (ints are valid floats).
|
||||
value.is_number().then_some(()).ok_or("number")
|
||||
}
|
||||
|
||||
// `Option<T>` accepts null or a value matching the inner type.
|
||||
TypeSchema::Option(inner) => {
|
||||
if value.is_null() {
|
||||
Ok(())
|
||||
} else {
|
||||
check_type(value, inner)
|
||||
}
|
||||
}
|
||||
|
||||
TypeSchema::Array(inner) => match value.as_array() {
|
||||
Some(items) => {
|
||||
for item in items {
|
||||
check_type(item, inner)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
None => Err("array"),
|
||||
},
|
||||
|
||||
TypeSchema::Enum { variants } => match value.as_str() {
|
||||
Some(s) if variants.contains(&s) => Ok(()),
|
||||
Some(_) => Err("one of the allowed enum variants"),
|
||||
None => Err("string"),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Attempts to invoke a registered RPC method by name.
|
||||
///
|
||||
/// Checks both the agent-facing controller registry and the internal-only registry,
|
||||
|
||||
@@ -361,6 +361,138 @@ fn validate_params_null_for_required_is_acceptable() {
|
||||
assert!(validate_params(&s, &p).is_ok());
|
||||
}
|
||||
|
||||
// --- validate_params type checking (C12) --------------------------------
|
||||
|
||||
#[test]
|
||||
fn validate_params_rejects_wrong_scalar_type() {
|
||||
let s = schema(
|
||||
"test",
|
||||
"fn",
|
||||
vec![FieldSchema {
|
||||
name: "count",
|
||||
ty: TypeSchema::U64,
|
||||
comment: "",
|
||||
required: true,
|
||||
}],
|
||||
);
|
||||
let mut p = Map::new();
|
||||
p.insert("count".into(), Value::String("nope".into()));
|
||||
let err = validate_params(&s, &p).unwrap_err();
|
||||
assert!(err.contains("invalid type for param 'count'"), "got: {err}");
|
||||
assert!(err.contains("expected unsigned integer"), "got: {err}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_params_accepts_correct_scalar_type() {
|
||||
let s = schema(
|
||||
"test",
|
||||
"fn",
|
||||
vec![FieldSchema {
|
||||
name: "flag",
|
||||
ty: TypeSchema::Bool,
|
||||
comment: "",
|
||||
required: true,
|
||||
}],
|
||||
);
|
||||
let mut p = Map::new();
|
||||
p.insert("flag".into(), Value::Bool(true));
|
||||
assert!(validate_params(&s, &p).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_params_validates_array_element_types() {
|
||||
let s = schema(
|
||||
"test",
|
||||
"fn",
|
||||
vec![FieldSchema {
|
||||
name: "ids",
|
||||
ty: TypeSchema::Array(Box::new(TypeSchema::String)),
|
||||
comment: "",
|
||||
required: true,
|
||||
}],
|
||||
);
|
||||
let mut ok = Map::new();
|
||||
ok.insert(
|
||||
"ids".into(),
|
||||
Value::Array(vec![Value::String("a".into()), Value::String("b".into())]),
|
||||
);
|
||||
assert!(validate_params(&s, &ok).is_ok());
|
||||
|
||||
let mut bad = Map::new();
|
||||
bad.insert(
|
||||
"ids".into(),
|
||||
Value::Array(vec![Value::String("a".into()), Value::Bool(true)]),
|
||||
);
|
||||
let err = validate_params(&s, &bad).unwrap_err();
|
||||
assert!(err.contains("invalid type for param 'ids'"), "got: {err}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_params_enforces_enum_variants() {
|
||||
let s = schema(
|
||||
"test",
|
||||
"fn",
|
||||
vec![FieldSchema {
|
||||
name: "mode",
|
||||
ty: TypeSchema::Enum {
|
||||
variants: vec!["read", "write"],
|
||||
},
|
||||
comment: "",
|
||||
required: true,
|
||||
}],
|
||||
);
|
||||
let mut ok = Map::new();
|
||||
ok.insert("mode".into(), Value::String("read".into()));
|
||||
assert!(validate_params(&s, &ok).is_ok());
|
||||
|
||||
let mut bad = Map::new();
|
||||
bad.insert("mode".into(), Value::String("delete".into()));
|
||||
let err = validate_params(&s, &bad).unwrap_err();
|
||||
assert!(err.contains("enum variants"), "got: {err}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_params_option_accepts_null_and_inner_type() {
|
||||
let s = schema(
|
||||
"test",
|
||||
"fn",
|
||||
vec![FieldSchema {
|
||||
name: "limit",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
|
||||
comment: "",
|
||||
required: false,
|
||||
}],
|
||||
);
|
||||
let mut null_p = Map::new();
|
||||
null_p.insert("limit".into(), Value::Null);
|
||||
assert!(validate_params(&s, &null_p).is_ok());
|
||||
|
||||
let mut val_p = Map::new();
|
||||
val_p.insert("limit".into(), Value::Number(5.into()));
|
||||
assert!(validate_params(&s, &val_p).is_ok());
|
||||
|
||||
let mut bad_p = Map::new();
|
||||
bad_p.insert("limit".into(), Value::String("x".into()));
|
||||
assert!(validate_params(&s, &bad_p).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_params_json_type_accepts_anything() {
|
||||
let s = schema(
|
||||
"test",
|
||||
"fn",
|
||||
vec![FieldSchema {
|
||||
name: "payload",
|
||||
ty: TypeSchema::Json,
|
||||
comment: "",
|
||||
required: true,
|
||||
}],
|
||||
);
|
||||
let mut p = Map::new();
|
||||
p.insert("payload".into(), Value::Array(vec![Value::Bool(true)]));
|
||||
assert!(validate_params(&s, &p).is_ok());
|
||||
}
|
||||
|
||||
// --- validate_registry edge cases ---------------------------------------
|
||||
|
||||
#[test]
|
||||
|
||||
+17
-8
@@ -34,8 +34,13 @@
|
||||
//! one-time login tokens, never raw session JWTs
|
||||
//! - `GET /auth/telegram` — external browser callback (carries its own token)
|
||||
//! - `GET /schema` — read-only schema discovery
|
||||
//! - `GET /events` — SSE stream; browser `EventSource` cannot set headers
|
||||
//! - `GET /ws/dictation` — WebSocket upgrade; browser WS API cannot set headers
|
||||
//! - `GET /events` — SSE stream; browser `EventSource` cannot set
|
||||
//! headers, so the handler enforces a bind-token /
|
||||
//! bearer credential itself
|
||||
//! - `GET /ws/dictation` — WebSocket upgrade; browser WS API cannot set
|
||||
//! headers, so the handler enforces the bearer
|
||||
//! (header or `?token=`) + origin itself before the
|
||||
//! upgrade (C4 / issue #1924)
|
||||
//! - `OPTIONS *` — CORS preflight (handled by outer CORS middleware)
|
||||
//!
|
||||
//! Endpoints that accept the bearer either via header **or** `?token=…` query
|
||||
@@ -71,10 +76,15 @@ static RPC_TOKEN: OnceLock<String> = OnceLock::new();
|
||||
|
||||
/// Paths that bypass bearer-token authentication.
|
||||
///
|
||||
/// `/rpc` and `/v1/*` carry executable surfaces and must be protected. All
|
||||
/// other routes are read-only, streaming, or WebSocket upgrades whose clients
|
||||
/// (browser `EventSource`, browser `WebSocket`) cannot set `Authorization`
|
||||
/// headers via standard APIs.
|
||||
/// `/rpc` and `/v1/*` carry executable surfaces and must be protected. The
|
||||
/// other routes are read-only, or are streaming / WebSocket upgrades whose
|
||||
/// clients (browser `EventSource`, browser `WebSocket`) cannot set
|
||||
/// `Authorization` headers via standard APIs. `/events` is not unauthenticated
|
||||
/// — it is exempt from the *middleware* header check but enforces its own
|
||||
/// bind-token credential inside the handler. `/ws/dictation` is NOT public: it
|
||||
/// is bearer-gated by this middleware via [`QUERY_TOKEN_PATHS`] (header or
|
||||
/// `?token=`) so an unauthenticated upgrade is rejected with 401 before the
|
||||
/// WebSocket handshake; the handler adds an origin check on top (finding C4).
|
||||
const PUBLIC_PATHS: &[&str] = &[
|
||||
"/",
|
||||
"/health",
|
||||
@@ -86,7 +96,6 @@ const PUBLIC_PATHS: &[&str] = &[
|
||||
"/oauth/mcp/callback",
|
||||
"/schema",
|
||||
"/events",
|
||||
"/ws/dictation",
|
||||
];
|
||||
|
||||
/// Paths that may authenticate via `?token=…` in the URL when no
|
||||
@@ -102,7 +111,7 @@ const PUBLIC_PATHS: &[&str] = &[
|
||||
/// Add new entries here only for SSE / WebSocket routes whose clients cannot
|
||||
/// send headers and that carry per-user data. The follow-up approvals stream
|
||||
/// (#1339) is the next planned addition.
|
||||
const QUERY_TOKEN_PATHS: &[&str] = &["/events/webhooks"];
|
||||
const QUERY_TOKEN_PATHS: &[&str] = &["/events/webhooks", "/ws/dictation"];
|
||||
|
||||
/// Operator-supplied environment variable that carries the RPC bearer in
|
||||
/// non-desktop deployments.
|
||||
|
||||
+103
-16
@@ -917,9 +917,84 @@ async fn desktop_auth_handler(
|
||||
)
|
||||
}
|
||||
|
||||
/// Query parameters for the dictation WebSocket endpoint.
|
||||
///
|
||||
/// Browser `WebSocket` cannot attach an `Authorization` header on upgrade, so
|
||||
/// the FE forwards the per-process core bearer as a `?token=…` query param —
|
||||
/// validated against the same in-process RPC token via [`verify_bearer_token`]
|
||||
/// (single source of truth, no separate credential).
|
||||
#[derive(Debug, serde::Deserialize)]
|
||||
struct DictationQuery {
|
||||
#[serde(default)]
|
||||
token: Option<String>,
|
||||
}
|
||||
|
||||
/// WebSocket upgrade handler for streaming voice dictation.
|
||||
async fn dictation_ws_handler(ws: WebSocketUpgrade) -> Response {
|
||||
///
|
||||
/// Authenticated before upgrade (C4 / issue #1924): the request must carry the
|
||||
/// per-process core bearer either as `Authorization: Bearer <token>` (CLI /
|
||||
/// native callers) or as `?token=<token>` (browser `WebSocket`, which cannot
|
||||
/// set headers), and — when an `Origin` header is present — that origin must be
|
||||
/// on the local-app allowlist, mirroring the Socket.IO handshake check. Missing
|
||||
/// or wrong credentials are rejected with 401 and the socket is never upgraded.
|
||||
async fn dictation_ws_handler(
|
||||
headers: axum::http::HeaderMap,
|
||||
Query(query): Query<DictationQuery>,
|
||||
ws: WebSocketUpgrade,
|
||||
) -> Response {
|
||||
log::info!("[ws] dictation WebSocket upgrade requested");
|
||||
|
||||
// Origin check (same allowlist Socket.IO enforces): native clients send no
|
||||
// Origin and are accepted; cross-origin browser pages are rejected even if
|
||||
// they somehow hold the bearer.
|
||||
let origin = headers
|
||||
.get(header::ORIGIN)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(str::trim);
|
||||
if !crate::core::socketio::origin_is_allowed(origin) {
|
||||
log::warn!("[ws] dictation upgrade rejected: disallowed origin {origin:?}");
|
||||
return (
|
||||
StatusCode::FORBIDDEN,
|
||||
Json(json!({
|
||||
"ok": false,
|
||||
"error": "forbidden",
|
||||
"message": "Origin not allowed for the dictation WebSocket."
|
||||
})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
|
||||
// Bearer check: header first, then `?token=` for browser WebSocket clients.
|
||||
let header_token = headers
|
||||
.get(header::AUTHORIZATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| v.strip_prefix("Bearer "))
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty());
|
||||
let bearer_ok = header_token
|
||||
.map(crate::core::auth::verify_bearer_token)
|
||||
.unwrap_or(false);
|
||||
let bearer_ok = bearer_ok
|
||||
|| query
|
||||
.token
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(crate::core::auth::verify_bearer_token)
|
||||
.unwrap_or(false);
|
||||
if !bearer_ok {
|
||||
log::warn!("[ws] dictation upgrade rejected: missing or invalid bearer token");
|
||||
return (
|
||||
StatusCode::UNAUTHORIZED,
|
||||
Json(json!({
|
||||
"ok": false,
|
||||
"error": "unauthorized",
|
||||
"message": "Missing or invalid token. Supply 'Authorization: Bearer <core>' or ?token=<core>."
|
||||
})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
|
||||
ws.on_upgrade(|socket| async move {
|
||||
let config = match crate::openhuman::config::rpc::load_config_with_timeout().await {
|
||||
Ok(c) => Arc::new(c),
|
||||
@@ -1713,25 +1788,37 @@ async fn run_server_inner(
|
||||
.ok()
|
||||
.filter(|s| !s.trim().is_empty())
|
||||
.is_some();
|
||||
// Auth subsystem was already initialised above; fall back to the live
|
||||
// token if neither input matched but somehow a token is seeded (e.g.
|
||||
// a future caller route that doesn't thread the value through here).
|
||||
let has_initialized_token = crate::core::auth::get_rpc_token()
|
||||
.map(|t| !t.trim().is_empty())
|
||||
.unwrap_or(false);
|
||||
let has_explicit_token = has_in_memory_token || has_env_token || has_initialized_token;
|
||||
// Fail closed (#1919): a non-loopback bind exposes the entire RPC
|
||||
// surface (tool execution, file access, credentials) to the network,
|
||||
// so it must be guarded by an OPERATOR-supplied bearer. Only an
|
||||
// explicit operator token counts here:
|
||||
// - in-memory handoff from the embedded caller (`rpc_token`), or
|
||||
// - `OPENHUMAN_CORE_TOKEN` in the process environment.
|
||||
// The self-generated `core.token` file (which `init_rpc_token` may
|
||||
// already have seeded into the auth subsystem above) does NOT satisfy
|
||||
// this requirement: remote clients cannot read that file, so treating
|
||||
// it as "explicit" would be fail-open. Refuse to bind instead of
|
||||
// serving an effectively unauthenticated network surface.
|
||||
let has_explicit_token = has_in_memory_token || has_env_token;
|
||||
if !has_explicit_token {
|
||||
log::error!(
|
||||
"[core] ⚠️ SECURITY WARNING: Binding on public address {resolved_host} without \
|
||||
an explicit OPENHUMAN_CORE_TOKEN. The RPC server will auto-generate a token, \
|
||||
but external clients will not know it. Set OPENHUMAN_CORE_TOKEN in your \
|
||||
.env file to secure the RPC endpoint."
|
||||
"[core] SECURITY: refusing to bind on public address {resolved_host} without an \
|
||||
explicit operator-supplied RPC token. Set {} in your environment (or hand the \
|
||||
bearer in-memory via the embedded core handle) to secure the RPC endpoint.",
|
||||
crate::core::auth::CORE_TOKEN_ENV_VAR
|
||||
);
|
||||
eprintln!(
|
||||
"\n\x1b[1;31m[SECURITY]\x1b[0m Binding on {resolved_host} without OPENHUMAN_CORE_TOKEN.\n\
|
||||
Set OPENHUMAN_CORE_TOKEN in .env to secure the RPC endpoint.\n\
|
||||
Without it, the auto-generated token is written to {{workspace}}/core.token\n\
|
||||
but remote clients will not be able to authenticate.\n"
|
||||
"\n\x1b[1;31m[SECURITY]\x1b[0m Refusing to bind on {resolved_host} without {}.\n\
|
||||
The auto-generated {{workspace}}/core.token does NOT secure a public bind —\n\
|
||||
remote clients cannot read it. Set {} in your environment to secure the\n\
|
||||
RPC endpoint, or bind on a loopback address.\n",
|
||||
crate::core::auth::CORE_TOKEN_ENV_VAR,
|
||||
crate::core::auth::CORE_TOKEN_ENV_VAR
|
||||
);
|
||||
anyhow::bail!(
|
||||
"refusing to bind on non-loopback address {resolved_host} without an explicit \
|
||||
operator-supplied RPC token ({})",
|
||||
crate::core::auth::CORE_TOKEN_ENV_VAR
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -47,7 +47,7 @@ struct HandshakeAuth {
|
||||
/// A missing `Origin` header is treated as a native (non-browser) client
|
||||
/// and accepted — only the cross-origin browser-page case is the targeted
|
||||
/// bad actor here.
|
||||
fn origin_is_allowed(origin: Option<&str>) -> bool {
|
||||
pub(crate) fn origin_is_allowed(origin: Option<&str>) -> bool {
|
||||
let Some(origin) = origin else {
|
||||
return true; // native clients (CLI, Tauri shell) — no Origin header
|
||||
};
|
||||
|
||||
@@ -238,7 +238,20 @@ impl Config {
|
||||
"Config loaded"
|
||||
);
|
||||
crate::openhuman::migrations::run_pending(&mut config).await;
|
||||
decrypt_config_secrets(&mut config, &openhuman_dir)?;
|
||||
let migrated_legacy_secrets = decrypt_config_secrets(&mut config, &openhuman_dir)?;
|
||||
if migrated_legacy_secrets {
|
||||
// One-time forced migration: a legacy `enc:` (XOR) secret was
|
||||
// upgraded to `enc2:` on read. Persist immediately so the
|
||||
// insecure ciphertext stops living on disk (audit C8). A save
|
||||
// failure is non-fatal — the config is still usable in memory
|
||||
// and migration will be retried on the next startup.
|
||||
if let Err(e) = config.save().await {
|
||||
log::warn!(
|
||||
"[security][config] failed to persist enc: -> enc2: secret migration; \
|
||||
will retry on next startup: {e}"
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(config)
|
||||
} else {
|
||||
let mut config = Config {
|
||||
@@ -304,7 +317,9 @@ impl Config {
|
||||
config.workspace_dir = workspace_dir;
|
||||
config.action_dir = resolve_action_dir(&config.action_dir_override);
|
||||
config.apply_env_overrides();
|
||||
decrypt_config_secrets(&mut config, &openhuman_dir)?;
|
||||
// Debug-dump path is read-only; ignore the migration signal (the
|
||||
// authoritative `load_or_init` path persists upgraded secrets).
|
||||
let _ = decrypt_config_secrets(&mut config, &openhuman_dir)?;
|
||||
Ok(config)
|
||||
}
|
||||
|
||||
|
||||
@@ -1,7 +1,29 @@
|
||||
use super::super::Config;
|
||||
use anyhow::{Context, Result};
|
||||
use std::path::Path;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
/// Set whenever a legacy `enc:` (XOR) secret is force-migrated to `enc2:`
|
||||
/// during a `decrypt_config_secrets` pass. The flag is scoped to a single pass:
|
||||
/// `decrypt_config_secrets` clears it on entry and reads (and clears) it on exit
|
||||
/// so the caller can persist the upgraded config (audit C8). The worst case
|
||||
/// under a (rare) concurrent load is an extra, idempotent `config.save()` — the
|
||||
/// re-encryption is a no-op once values are already `enc2:`, so an occasional
|
||||
/// spurious save is harmless and never loses data.
|
||||
fn migration_flag() -> &'static AtomicBool {
|
||||
static MIGRATED: AtomicBool = AtomicBool::new(false);
|
||||
&MIGRATED
|
||||
}
|
||||
|
||||
/// Decrypt one optional secret in place.
|
||||
///
|
||||
/// When the stored value uses the legacy, insecure `enc:` (XOR) format this
|
||||
/// performs a **forced one-time migration**: the field is decrypted via
|
||||
/// `decrypt_and_migrate` (which rewrites it to the `enc2:` / ChaCha20-Poly1305
|
||||
/// format) and the process-wide migration flag is raised so the caller can
|
||||
/// persist the upgraded config back to disk. This stops legacy XOR ciphertext
|
||||
/// from persisting indefinitely (see audit C8). Reading existing `enc:` values
|
||||
/// still succeeds throughout the migration.
|
||||
fn decrypt_optional_secret(
|
||||
store: &crate::openhuman::keyring::SecretStore,
|
||||
value: &mut Option<String>,
|
||||
@@ -9,8 +31,19 @@ fn decrypt_optional_secret(
|
||||
) -> Result<()> {
|
||||
if let Some(raw) = value.clone() {
|
||||
if crate::openhuman::keyring::SecretStore::is_encrypted(&raw) {
|
||||
match store.decrypt(&raw) {
|
||||
Ok(plaintext) => *value = Some(plaintext),
|
||||
// Legacy `enc:` values are migrated to `enc2:` on read so they are
|
||||
// rewritten on the next config save instead of lingering forever.
|
||||
match store.decrypt_and_migrate(&raw) {
|
||||
Ok((plaintext, migrated)) => {
|
||||
if migrated.is_some() {
|
||||
log::warn!(
|
||||
"[security][config] migrated legacy enc: secret '{field_name}' \
|
||||
to enc2: (ChaCha20-Poly1305); config will be re-saved to persist"
|
||||
);
|
||||
migration_flag().store(true, Ordering::Relaxed);
|
||||
}
|
||||
*value = Some(plaintext);
|
||||
}
|
||||
Err(e) => {
|
||||
// Decryption key is inaccessible (e.g. rotated, keyring reset, or
|
||||
// migrated across machines). Clear the field so config loads
|
||||
@@ -54,10 +87,17 @@ fn encrypt_optional_secret(
|
||||
/// Called during config load when `secrets.encrypt` is true. Only decrypts
|
||||
/// values that have the `enc:` or `enc2:` prefix; plaintext values are
|
||||
/// returned as-is. This is a no-op when encryption is disabled.
|
||||
pub(super) fn decrypt_config_secrets(config: &mut Config, openhuman_dir: &Path) -> Result<()> {
|
||||
///
|
||||
/// Returns `true` if any field used the legacy, insecure `enc:` format and was
|
||||
/// force-migrated to `enc2:` during this pass. The caller should persist the
|
||||
/// config (e.g. `config.save()`) when this is `true` so the upgraded ciphertext
|
||||
/// is written to disk and the legacy XOR value stops persisting (audit C8).
|
||||
pub(super) fn decrypt_config_secrets(config: &mut Config, openhuman_dir: &Path) -> Result<bool> {
|
||||
if !config.secrets.encrypt {
|
||||
return Ok(());
|
||||
return Ok(false);
|
||||
}
|
||||
// Reset the per-pass migration flag before decrypting any field.
|
||||
migration_flag().store(false, Ordering::Relaxed);
|
||||
let store = crate::openhuman::keyring::SecretStore::new(openhuman_dir, true);
|
||||
|
||||
decrypt_optional_secret(&store, &mut config.api_key, "api_key")?;
|
||||
@@ -145,7 +185,7 @@ pub(super) fn decrypt_config_secrets(config: &mut Config, openhuman_dir: &Path)
|
||||
qq.app_secret = tok.unwrap_or_default();
|
||||
}
|
||||
|
||||
Ok(())
|
||||
Ok(migration_flag().swap(false, Ordering::Relaxed))
|
||||
}
|
||||
|
||||
/// Encrypt all secret fields in the configuration before writing to disk.
|
||||
|
||||
@@ -8,6 +8,30 @@ use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Extract an HTTP `4xx` status code from an error message, but only when it
|
||||
/// appears in a *structured* position — never from arbitrary digit runs in
|
||||
/// free text (audit C10). Recognised positions:
|
||||
///
|
||||
/// - the documented provider envelope `"… API error (<status>): …"`
|
||||
/// (e.g. `"OpenAI API error (401 Unauthorized): …"`),
|
||||
/// - an explicit `HTTP <status>` marker,
|
||||
/// - a `status: <status>` / `status <status>` field,
|
||||
/// - a status code that *leads* the message (e.g. `"404 Not Found"`).
|
||||
///
|
||||
/// Returns the matched code (always in `400..=499`) or `None`.
|
||||
fn structured_http_4xx(msg: &str) -> Option<u16> {
|
||||
static RE: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
|
||||
let re = RE.get_or_init(|| {
|
||||
// (?i) case-insensitive; capture the 4xx in any one of the structured
|
||||
// anchors. `\A` matches start-of-string for the leading-status form.
|
||||
regex::Regex::new(r"(?i)(?:\(|HTTP\s+|status[:\s]+|\A)(4\d\d)\b")
|
||||
.expect("static is_non_retryable 4xx regex is valid")
|
||||
});
|
||||
re.captures(msg)
|
||||
.and_then(|c| c.get(1))
|
||||
.and_then(|m| m.as_str().parse::<u16>().ok())
|
||||
}
|
||||
|
||||
/// Check if an error is non-retryable (client errors that won't resolve with retries).
|
||||
fn is_non_retryable(err: &anyhow::Error) -> bool {
|
||||
if is_context_window_exceeded(err) {
|
||||
@@ -27,12 +51,14 @@ fn is_non_retryable(err: &anyhow::Error) -> bool {
|
||||
return status.is_client_error() && code != 429 && code != 408;
|
||||
}
|
||||
}
|
||||
for word in msg.split(|c: char| !c.is_ascii_digit()) {
|
||||
if let Ok(code) = word.parse::<u16>() {
|
||||
if (400..500).contains(&code) {
|
||||
return code != 429 && code != 408;
|
||||
}
|
||||
}
|
||||
// Don't infer an HTTP status from *any* free-text digit run — strings like
|
||||
// "took 450ms" (→450) or model ids like "gpt-4-0409" (→409) would be
|
||||
// misclassified as permanent client errors and short-circuit retries
|
||||
// (audit C10). Match only a 4xx code in a *structured* position: the
|
||||
// documented `… API error (<status>): …` envelope, an `HTTP <status>` /
|
||||
// `status: <status>` marker, or a status that leads the message.
|
||||
if let Some(code) = structured_http_4xx(&msg) {
|
||||
return code != 429 && code != 408;
|
||||
}
|
||||
|
||||
let msg_lower = msg.to_lowercase();
|
||||
@@ -330,7 +356,13 @@ fn format_failure_aggregate(
|
||||
|
||||
/// Provider wrapper with retry, fallback, auth rotation, and model failover.
|
||||
pub struct ReliableProvider {
|
||||
providers: Vec<(String, Box<dyn Provider>)>,
|
||||
/// Stored behind `Arc` (not `Box`) so the streaming failover path can hand
|
||||
/// owned, `'static` provider handles to the consumer task and create
|
||||
/// candidate streams *lazily* — issuing each upstream request only when the
|
||||
/// previous candidate has actually failed (see `stream_chat_with_system`).
|
||||
/// The public `new` constructor still accepts `Box<dyn Provider>`; the
|
||||
/// conversion happens internally so callers are unaffected.
|
||||
providers: Vec<(String, std::sync::Arc<dyn Provider>)>,
|
||||
max_retries: u32,
|
||||
base_backoff_ms: u64,
|
||||
/// Extra API keys for rotation (index tracks round-robin position).
|
||||
@@ -347,7 +379,10 @@ impl ReliableProvider {
|
||||
base_backoff_ms: u64,
|
||||
) -> Self {
|
||||
Self {
|
||||
providers,
|
||||
providers: providers
|
||||
.into_iter()
|
||||
.map(|(name, p)| (name, std::sync::Arc::from(p)))
|
||||
.collect(),
|
||||
max_retries,
|
||||
base_backoff_ms: base_backoff_ms.max(50),
|
||||
api_keys: Vec::new(),
|
||||
@@ -1006,25 +1041,27 @@ impl Provider for ReliableProvider {
|
||||
let model_chain: Vec<String> = models.into_iter().map(|m| m.to_string()).collect();
|
||||
let base_backoff_ms = self.base_backoff_ms;
|
||||
|
||||
// Collect provider streams lazily inside the task — we need owned data
|
||||
// Provider trait is object-safe, so we call stream_chat_with_system per attempt
|
||||
// We need to pre-create all possible streams since Provider is behind &self
|
||||
// Instead, collect the streams for each provider+model combo upfront
|
||||
let mut candidate_streams: Vec<(
|
||||
String,
|
||||
String,
|
||||
stream::BoxStream<'static, StreamResult<StreamChunk>>,
|
||||
)> = Vec::new();
|
||||
// Capture only owned `(provider_name, provider, model)` *tuples* up-front
|
||||
// — NOT the streams themselves. The provider impl spawns the upstream
|
||||
// HTTP POST the instant a stream is created, so eagerly building the
|
||||
// full provider×model product here would fire every fallback request
|
||||
// at once (duplicate billing/side-effects). Instead we clone the
|
||||
// `Arc<dyn Provider>` handles and call `stream_chat_with_system` lazily
|
||||
// inside the consumer task, immediately before each candidate is tried
|
||||
// — mirroring the sequential non-streaming paths. (audit C2)
|
||||
//
|
||||
// `system_prompt` / `message` are borrowed from the caller and the
|
||||
// spawned task is `'static`, so own them here.
|
||||
let system_prompt_owned: Option<String> = system_prompt.map(|s| s.to_string());
|
||||
let message_owned: String = message.to_string();
|
||||
let mut candidates: Vec<(String, std::sync::Arc<dyn Provider>, String)> = Vec::new();
|
||||
for current_model in &model_chain {
|
||||
for (provider_name, provider) in &streaming_providers {
|
||||
let s = provider.stream_chat_with_system(
|
||||
system_prompt,
|
||||
message,
|
||||
current_model,
|
||||
temperature,
|
||||
options,
|
||||
);
|
||||
candidate_streams.push(((*provider_name).clone(), current_model.clone(), s));
|
||||
candidates.push((
|
||||
(*provider_name).clone(),
|
||||
std::sync::Arc::clone(provider),
|
||||
current_model.clone(),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1032,11 +1069,24 @@ impl Provider for ReliableProvider {
|
||||
let max_retries = self.max_retries;
|
||||
|
||||
tokio::spawn(async move {
|
||||
for (provider_name, current_model, mut candidate_stream) in candidate_streams {
|
||||
for (provider_name, provider, current_model) in candidates {
|
||||
let mut backoff_ms = base_backoff_ms;
|
||||
let mut attempts = 0u32;
|
||||
|
||||
loop {
|
||||
// Create (and thereby fire) the candidate stream lazily here,
|
||||
// immediately before we attempt it. On a retryable failure we
|
||||
// re-create it on the next loop iteration rather than
|
||||
// re-polling the previous, already-exhausted stream (which
|
||||
// only yields `None` after its single error). (audit C2/C6)
|
||||
let mut candidate_stream = provider.stream_chat_with_system(
|
||||
system_prompt_owned.as_deref(),
|
||||
&message_owned,
|
||||
¤t_model,
|
||||
temperature,
|
||||
options,
|
||||
);
|
||||
|
||||
match candidate_stream.next().await {
|
||||
Some(Ok(chunk)) => {
|
||||
// First chunk succeeded — commit to this stream
|
||||
@@ -1069,7 +1119,8 @@ impl Provider for ReliableProvider {
|
||||
attempts += 1;
|
||||
tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
|
||||
backoff_ms = (backoff_ms.saturating_mul(2)).min(10_000);
|
||||
// Continue inner loop — stream may yield more items
|
||||
// Re-create the candidate stream on the next iteration.
|
||||
continue;
|
||||
}
|
||||
None => {
|
||||
// Stream exhausted without success
|
||||
|
||||
@@ -221,6 +221,40 @@ fn non_retryable_detects_common_patterns() {
|
||||
)));
|
||||
}
|
||||
|
||||
// C10: a 4xx-looking digit run that appears in *free text* (latency figures,
|
||||
// model ids, token counts) must NOT be inferred as a permanent HTTP client
|
||||
// error — that wrongly short-circuits retries/fallback for transient failures.
|
||||
#[test]
|
||||
fn non_retryable_ignores_free_text_digit_runs() {
|
||||
// "450" here is a latency figure, not a 450 status.
|
||||
assert!(
|
||||
!is_non_retryable(&anyhow::anyhow!("upstream took 450ms to respond, retrying")),
|
||||
"latency figures must not be read as an HTTP status"
|
||||
);
|
||||
// "0409" embedded in a model id used to scan to 409.
|
||||
assert!(
|
||||
!is_non_retryable(&anyhow::anyhow!("gpt-4-0409 returned an empty completion")),
|
||||
"model-id digits must not be read as an HTTP status"
|
||||
);
|
||||
// A bare 4xx-shaped token mid-sentence (not in a structured position) is
|
||||
// also ignored now.
|
||||
assert!(
|
||||
!is_non_retryable(&anyhow::anyhow!(
|
||||
"received 412 partial bytes before connection reset"
|
||||
)),
|
||||
"mid-text digit runs must not be read as an HTTP status"
|
||||
);
|
||||
// Sanity: the structured envelope is still classified as non-retryable.
|
||||
assert!(
|
||||
is_non_retryable(&anyhow::anyhow!(
|
||||
"custom_openai API error (403 Forbidden): nope"
|
||||
)),
|
||||
"the documented (<status>) envelope must still be detected"
|
||||
);
|
||||
// Sanity: a leading status (no envelope) is still detected.
|
||||
assert!(is_non_retryable(&anyhow::anyhow!("404 Not Found")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn context_window_error_aborts_retries_and_model_fallbacks() {
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
@@ -418,6 +452,161 @@ async fn session_expired_aborts_retries_streaming() {
|
||||
);
|
||||
}
|
||||
|
||||
/// Streaming mock whose stream fails with a *retryable* `StreamError` for the
|
||||
/// first `fail_until` creations and then yields a single successful chunk. Each
|
||||
/// stream is mpsc-like (exhausts after one item), exactly like the real
|
||||
/// provider impl — so a retry that re-polls the same dead stream would see
|
||||
/// `None` and give up. `stream_calls` records how many times a stream was
|
||||
/// created, the signal used to prove lazy creation (audit C2) and
|
||||
/// recreate-on-retry (audit C6).
|
||||
struct StreamingRetryMock {
|
||||
stream_calls: Arc<AtomicUsize>,
|
||||
fail_until: usize,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Provider for StreamingRetryMock {
|
||||
async fn chat_with_system(
|
||||
&self,
|
||||
_system_prompt: Option<&str>,
|
||||
_message: &str,
|
||||
_model: &str,
|
||||
_temperature: f64,
|
||||
) -> anyhow::Result<String> {
|
||||
anyhow::bail!("unused")
|
||||
}
|
||||
|
||||
async fn chat_with_history(
|
||||
&self,
|
||||
_messages: &[ChatMessage],
|
||||
_model: &str,
|
||||
_temperature: f64,
|
||||
) -> anyhow::Result<String> {
|
||||
anyhow::bail!("unused")
|
||||
}
|
||||
|
||||
fn supports_streaming(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn stream_chat_with_system(
|
||||
&self,
|
||||
_system_prompt: Option<&str>,
|
||||
_message: &str,
|
||||
_model: &str,
|
||||
_temperature: f64,
|
||||
_options: StreamOptions,
|
||||
) -> futures_util::stream::BoxStream<'static, StreamResult<StreamChunk>> {
|
||||
use futures_util::{stream, StreamExt};
|
||||
// The Nth stream creation (1-based) fails if N <= fail_until, else
|
||||
// succeeds. Firing the HTTP request happens *here*, so counting
|
||||
// creations is the proxy for "requests issued".
|
||||
let n = self.stream_calls.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
let succeed = n > self.fail_until;
|
||||
stream::once(async move {
|
||||
if succeed {
|
||||
Ok(StreamChunk::delta("hello"))
|
||||
} else {
|
||||
// A generic provider error — retryable per
|
||||
// is_stream_error_non_retryable.
|
||||
Err(StreamError::Provider("transient upstream blip".to_string()))
|
||||
}
|
||||
})
|
||||
.boxed()
|
||||
}
|
||||
}
|
||||
|
||||
/// C2: streaming failover must NOT pre-fire every provider×model stream. With a
|
||||
/// 2-model fallback chain and a provider that succeeds on the first attempt,
|
||||
/// only ONE stream may be created (the winning candidate) — not the full
|
||||
/// cartesian product.
|
||||
#[tokio::test]
|
||||
async fn streaming_does_not_prefire_all_candidates() {
|
||||
use futures_util::StreamExt;
|
||||
|
||||
let stream_calls = Arc::new(AtomicUsize::new(0));
|
||||
let mut fallbacks = HashMap::new();
|
||||
fallbacks.insert(
|
||||
"model-a".to_string(),
|
||||
vec!["model-b".to_string(), "model-c".to_string()],
|
||||
);
|
||||
|
||||
let provider = ReliableProvider::new(
|
||||
vec![(
|
||||
"p".into(),
|
||||
Box::new(StreamingRetryMock {
|
||||
stream_calls: Arc::clone(&stream_calls),
|
||||
fail_until: 0, // succeed immediately
|
||||
}),
|
||||
)],
|
||||
3,
|
||||
1,
|
||||
)
|
||||
.with_model_fallbacks(fallbacks);
|
||||
|
||||
let mut stream =
|
||||
provider.stream_chat_with_system(None, "hi", "model-a", 0.0, StreamOptions::new(true));
|
||||
|
||||
let mut chunks = Vec::new();
|
||||
while let Some(item) = stream.next().await {
|
||||
if let Ok(chunk) = item {
|
||||
chunks.push(chunk.delta);
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(chunks, vec!["hello".to_string()]);
|
||||
assert_eq!(
|
||||
stream_calls.load(Ordering::SeqCst),
|
||||
1,
|
||||
"only the winning candidate may create a stream; the rest must stay lazy (C2)"
|
||||
);
|
||||
}
|
||||
|
||||
/// C6: a retryable streaming failure must RE-CREATE the candidate stream on the
|
||||
/// next attempt rather than re-poll the already-exhausted one. With
|
||||
/// `fail_until = 2` and `max_retries = 3`, the same provider/model is attempted
|
||||
/// up to 4 times; creations 1 and 2 fail, creation 3 succeeds. If the retry
|
||||
/// loop re-polled the dead stream instead of recreating it, we'd only ever see
|
||||
/// a single creation and the call would fail.
|
||||
#[tokio::test]
|
||||
async fn streaming_retry_recreates_stream() {
|
||||
use futures_util::StreamExt;
|
||||
|
||||
let stream_calls = Arc::new(AtomicUsize::new(0));
|
||||
let provider = ReliableProvider::new(
|
||||
vec![(
|
||||
"p".into(),
|
||||
Box::new(StreamingRetryMock {
|
||||
stream_calls: Arc::clone(&stream_calls),
|
||||
fail_until: 2,
|
||||
}),
|
||||
)],
|
||||
3,
|
||||
1,
|
||||
);
|
||||
|
||||
let mut stream =
|
||||
provider.stream_chat_with_system(None, "hi", "reasoning-v1", 0.0, StreamOptions::new(true));
|
||||
|
||||
let mut chunks = Vec::new();
|
||||
while let Some(item) = stream.next().await {
|
||||
if let Ok(chunk) = item {
|
||||
chunks.push(chunk.delta);
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
chunks,
|
||||
vec!["hello".to_string()],
|
||||
"retry must eventually recover once the recreated stream succeeds (C6)"
|
||||
);
|
||||
assert_eq!(
|
||||
stream_calls.load(Ordering::SeqCst),
|
||||
3,
|
||||
"each retry attempt must recreate the candidate stream (C6), not re-poll the dead one"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn aggregated_error_marks_non_retryable_model_mismatch_with_details() {
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
@@ -16,11 +16,15 @@
|
||||
//! # Security notes
|
||||
//!
|
||||
//! ## Authentication
|
||||
//! `GET /ws/dictation` is intentionally exempt from Bearer-token authentication because
|
||||
//! the browser WebSocket API cannot set arbitrary request headers on upgrade. The correct
|
||||
//! auth mechanism is a separate maintainer decision; see `src/core/auth.rs` for the
|
||||
//! documented exemption. Do NOT add a Bearer-header check here — it will not work from
|
||||
//! browsers and the design decision is tracked in issue #1924.
|
||||
//! `GET /ws/dictation` is authenticated at the upgrade boundary (C4 / issue #1924).
|
||||
//! The browser WebSocket API cannot set arbitrary request headers on upgrade, so the
|
||||
//! check lives in `dictation_ws_handler` (`src/core/jsonrpc.rs`), not here: it requires
|
||||
//! the per-process core bearer via `Authorization: Bearer <token>` (native callers) or
|
||||
//! `?token=<token>` (browser clients), plus the same origin allowlist Socket.IO enforces,
|
||||
//! and rejects the upgrade with 401/403 before this function runs. Do NOT add a
|
||||
//! Bearer-header check in this function — by the time `handle_dictation_ws` is reached the
|
||||
//! upgrade has already been authenticated, and a header check here would not work from
|
||||
//! browsers anyway.
|
||||
//!
|
||||
//! ## Memory cap
|
||||
//! The full-audio accumulation buffer (`full_audio_buf`) is bounded by
|
||||
|
||||
@@ -20,8 +20,11 @@
|
||||
// For sovereign users who prefer plaintext, `secrets.encrypt = false` disables this.
|
||||
//
|
||||
// Migration: values with the legacy `enc:` prefix (XOR cipher) are decrypted
|
||||
// using the old algorithm for backward compatibility. New encryptions always
|
||||
// produce `enc2:` (ChaCha20-Poly1305).
|
||||
// using the old algorithm for backward compatibility, but the config loader
|
||||
// force-migrates them to `enc2:` on read (`decrypt_and_migrate`) and persists
|
||||
// the upgrade so the insecure ciphertext never lingers on disk (audit C8). The
|
||||
// legacy decode path logs a loud security warning every time it runs. New
|
||||
// encryptions always produce `enc2:` (ChaCha20-Poly1305).
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use chacha20poly1305::aead::{Aead, KeyInit, OsRng};
|
||||
@@ -31,6 +34,7 @@ use std::fs;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
use std::time::Duration;
|
||||
use zeroize::Zeroizing;
|
||||
|
||||
use super::crypto;
|
||||
|
||||
@@ -160,7 +164,22 @@ impl SecretStore {
|
||||
}
|
||||
|
||||
/// Decrypt using legacy XOR cipher (insecure, for backward compatibility only).
|
||||
///
|
||||
/// This is the single choke-point for the insecure `enc:` format, so it logs
|
||||
/// a loud security warning every time it runs. The repeating-key XOR is
|
||||
/// unauthenticated and leaks the master key to anyone who knows (or can guess
|
||||
/// the structure of) the plaintext — e.g. `xoxb-`/`sk-` token prefixes give
|
||||
/// `key[i] = ct[i] XOR pt[i]`. Callers on the read path should prefer
|
||||
/// [`decrypt_and_migrate`](Self::decrypt_and_migrate) so the value is rewritten
|
||||
/// to `enc2:` and the legacy ciphertext stops persisting on disk.
|
||||
fn decrypt_legacy_xor(&self, hex_str: &str) -> Result<String> {
|
||||
log::warn!(
|
||||
"[security] decrypting legacy XOR-encrypted secret (enc: prefix). \
|
||||
This format is INSECURE (unauthenticated repeating-key XOR; known-plaintext \
|
||||
recovers the master key) and must be migrated to enc2: (ChaCha20-Poly1305). \
|
||||
It will be removed in a future release — re-save your config or settings to \
|
||||
force migration."
|
||||
);
|
||||
let ciphertext = hex_decode(hex_str)
|
||||
.context("Failed to decode legacy encrypted secret (corrupt hex)")?;
|
||||
let key = self.load_or_create_key()?;
|
||||
@@ -187,7 +206,7 @@ impl SecretStore {
|
||||
keyring_user_id_from_dir(self.key_path.parent().unwrap_or_else(|| Path::new(".")))
|
||||
}
|
||||
|
||||
fn load_or_create_key_from_keyring(&self) -> Result<Vec<u8>> {
|
||||
fn load_or_create_key_from_keyring(&self) -> Result<Zeroizing<Vec<u8>>> {
|
||||
let user_id = self.keyring_user_id();
|
||||
match crate::openhuman::keyring::migrate_from_file(
|
||||
&user_id,
|
||||
@@ -228,7 +247,11 @@ impl SecretStore {
|
||||
/// The decoded key is cached process-wide keyed by `key_path`, so repeated
|
||||
/// callers (e.g. every `app_state_snapshot` poll) hit memory instead of
|
||||
/// disk/keychain lookup.
|
||||
fn load_or_create_key(&self) -> Result<Vec<u8>> {
|
||||
///
|
||||
/// The key bytes are wrapped in [`Zeroizing`] so every copy — the returned
|
||||
/// value, the cache entry, and any intermediate buffers — is wiped from
|
||||
/// memory on drop rather than lingering in the heap/swap/core-dumps.
|
||||
fn load_or_create_key(&self) -> Result<Zeroizing<Vec<u8>>> {
|
||||
// Normalize the path once so all callers share the same cache slot
|
||||
// regardless of how `key_path` was spelled (relative vs absolute,
|
||||
// symlinks, case-variants on Windows).
|
||||
@@ -462,18 +485,20 @@ fn normalize_cache_path(path: &Path) -> PathBuf {
|
||||
/// transiently inaccessible (AV scanners holding a handle), and re-reading
|
||||
/// turned that transient failure into a perma-failure for every subsequent
|
||||
/// RPC call.
|
||||
fn key_cache() -> &'static Mutex<HashMap<PathBuf, Vec<u8>>> {
|
||||
static CACHE: OnceLock<Mutex<HashMap<PathBuf, Vec<u8>>>> = OnceLock::new();
|
||||
fn key_cache() -> &'static Mutex<HashMap<PathBuf, Zeroizing<Vec<u8>>>> {
|
||||
static CACHE: OnceLock<Mutex<HashMap<PathBuf, Zeroizing<Vec<u8>>>>> = OnceLock::new();
|
||||
CACHE.get_or_init(|| Mutex::new(HashMap::new()))
|
||||
}
|
||||
|
||||
fn cached_key(path: &Path) -> Option<Vec<u8>> {
|
||||
fn cached_key(path: &Path) -> Option<Zeroizing<Vec<u8>>> {
|
||||
key_cache().lock().ok()?.get(path).cloned()
|
||||
}
|
||||
|
||||
fn cache_key(path: &Path, key: &[u8]) {
|
||||
if let Ok(mut cache) = key_cache().lock() {
|
||||
cache.insert(path.to_path_buf(), key.to_vec());
|
||||
// Stored as `Zeroizing` so the cached copy is wiped from memory when the
|
||||
// entry is removed (e.g. via `clear_cached_key`) or the process exits.
|
||||
cache.insert(path.to_path_buf(), Zeroizing::new(key.to_vec()));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -621,12 +646,12 @@ fn xor_cipher(data: &[u8], key: &[u8]) -> Vec<u8> {
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn generate_random_key() -> Vec<u8> {
|
||||
crypto::generate_random_bytes(KEY_LEN)
|
||||
fn generate_random_key() -> Zeroizing<Vec<u8>> {
|
||||
Zeroizing::new(crypto::generate_random_bytes(KEY_LEN))
|
||||
}
|
||||
|
||||
fn decode_key_hex(hex_key: &str) -> Result<Vec<u8>> {
|
||||
let key = hex_decode(hex_key).context("Secret key file is corrupt")?;
|
||||
fn decode_key_hex(hex_key: &str) -> Result<Zeroizing<Vec<u8>>> {
|
||||
let key = Zeroizing::new(hex_decode(hex_key).context("Secret key file is corrupt")?);
|
||||
anyhow::ensure!(
|
||||
key.len() == KEY_LEN,
|
||||
"Secret key file has wrong length: expected {KEY_LEN} bytes, got {}",
|
||||
|
||||
@@ -784,6 +784,26 @@ fn generate_random_key_correct_length() {
|
||||
assert_eq!(key.len(), KEY_LEN);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn master_key_is_zeroizing_and_cache_stable() {
|
||||
// Regression for audit C9: the master key is returned wrapped in
|
||||
// `zeroize::Zeroizing` (wiped on drop) and the cached copy is the same
|
||||
// bytes on a subsequent load (cache hit). The static type assertion below
|
||||
// fails to compile if the return type stops being `Zeroizing`.
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let store = SecretStore::new(tmp.path(), true);
|
||||
|
||||
let key: zeroize::Zeroizing<Vec<u8>> = store.load_or_create_key().unwrap();
|
||||
assert_eq!(key.len(), KEY_LEN);
|
||||
|
||||
// Second load must return identical key bytes from the process-wide cache.
|
||||
let key2 = store.load_or_create_key().unwrap();
|
||||
assert_eq!(
|
||||
&*key, &*key2,
|
||||
"cached master key must be stable across loads"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn generate_random_key_not_all_zeros() {
|
||||
let key = generate_random_key();
|
||||
|
||||
@@ -116,6 +116,9 @@ struct CircuitBreaker {
|
||||
consecutive_failures: AtomicU32,
|
||||
tripped: AtomicBool,
|
||||
last_trip: PMutex<Option<Instant>>,
|
||||
/// Set once a `SystemStartup` mark has been published for this path so a
|
||||
/// fresh boot reports a real status without re-emitting on every open.
|
||||
startup_emitted: AtomicBool,
|
||||
}
|
||||
|
||||
impl CircuitBreaker {
|
||||
@@ -124,13 +127,27 @@ impl CircuitBreaker {
|
||||
consecutive_failures: AtomicU32::new(0),
|
||||
tripped: AtomicBool::new(false),
|
||||
last_trip: PMutex::new(None),
|
||||
startup_emitted: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
|
||||
fn record_success(&self) {
|
||||
/// Records a successful init. Returns `true` if this call cleared a
|
||||
/// previously-tripped breaker (i.e. a transition back to healthy that the
|
||||
/// caller should announce on the bus). Returns `false` for the steady-state
|
||||
/// case where the breaker was already untripped, so we don't spam a
|
||||
/// `HealthChanged{healthy:true}` event on every successful call.
|
||||
fn record_success(&self) -> bool {
|
||||
self.consecutive_failures.store(0, Ordering::Relaxed);
|
||||
self.tripped.store(false, Ordering::Relaxed);
|
||||
*self.last_trip.lock() = None;
|
||||
// `swap` reports the prior value: `true` means we just transitioned
|
||||
// from tripped → untripped, which is the recovery edge to announce.
|
||||
self.tripped.swap(false, Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// Returns `true` exactly once per breaker — on the first successful open —
|
||||
/// so the caller emits a single `SystemStartup` mark for this path.
|
||||
fn mark_startup_emitted(&self) -> bool {
|
||||
!self.startup_emitted.swap(true, Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// Records one more failure. Returns `true` if this call just tripped the
|
||||
@@ -447,9 +464,42 @@ pub(crate) fn get_or_init_connection(config: &Config) -> Result<Arc<PMutex<Conne
|
||||
.connections
|
||||
.lock()
|
||||
.insert(db_path.clone(), Arc::clone(&arc_conn));
|
||||
// Reset any prior failure counter now that init succeeded.
|
||||
if let Some(breaker) = conn_cache().breakers.lock().get(&db_path) {
|
||||
breaker.record_success();
|
||||
// Reset any prior failure counter now that init succeeded. Use (or
|
||||
// lazily create) the persistent breaker so a clean first boot still
|
||||
// has somewhere to record the one-shot startup mark.
|
||||
let breaker = {
|
||||
let mut guard = conn_cache().breakers.lock();
|
||||
guard
|
||||
.entry(db_path.clone())
|
||||
.or_insert_with(|| Arc::new(CircuitBreaker::new()))
|
||||
.clone()
|
||||
};
|
||||
// Emit a one-time `SystemStartup` so a fresh boot reports a real
|
||||
// status for `memory_tree_db` instead of "unknown" until the first
|
||||
// failure. Fires once per path for the process lifetime.
|
||||
if breaker.mark_startup_emitted() {
|
||||
let _ = crate::core::event_bus::publish_global(
|
||||
crate::core::event_bus::DomainEvent::SystemStartup {
|
||||
component: "memory_tree_db".to_string(),
|
||||
},
|
||||
);
|
||||
}
|
||||
// Only announce recovery on the transition back to healthy — i.e.
|
||||
// when the breaker had previously tripped (driving `/health` to a
|
||||
// permanent 503). Steady-state successes stay silent so we don't
|
||||
// spam a `HealthChanged{healthy:true}` event on every call.
|
||||
if breaker.record_success() {
|
||||
log::info!(
|
||||
"[memory_tree] circuit breaker recovered for {}: DB init succeeded after a prior trip",
|
||||
db_path.display()
|
||||
);
|
||||
let _ = crate::core::event_bus::publish_global(
|
||||
crate::core::event_bus::DomainEvent::HealthChanged {
|
||||
component: "memory_tree_db".to_string(),
|
||||
healthy: true,
|
||||
message: None,
|
||||
},
|
||||
);
|
||||
}
|
||||
log::debug!("[memory_tree] DB connection cached and ready");
|
||||
Ok(arc_conn)
|
||||
@@ -546,3 +596,42 @@ pub fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result
|
||||
let guard = conn_arc.lock();
|
||||
f(&guard)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// `record_success` must only report a recovery transition (`true`) when
|
||||
/// it actually clears a tripped breaker — the signal `get_or_init_connection`
|
||||
/// uses to publish `HealthChanged{healthy:true}` exactly once instead of on
|
||||
/// every successful call (C1).
|
||||
#[test]
|
||||
fn record_success_announces_only_on_trip_to_healthy_transition() {
|
||||
let cb = CircuitBreaker::new();
|
||||
|
||||
// Untripped breaker: a success is steady-state, not a transition.
|
||||
assert!(!cb.record_success());
|
||||
|
||||
// Trip the breaker by crossing the failure threshold.
|
||||
let mut tripped = false;
|
||||
for _ in 0..CB_THRESHOLD {
|
||||
tripped = cb.record_failure();
|
||||
}
|
||||
assert!(tripped, "breaker should trip at CB_THRESHOLD failures");
|
||||
|
||||
// First success after a trip is the recovery edge → announce once.
|
||||
assert!(cb.record_success());
|
||||
// Subsequent successes are steady-state → stay silent.
|
||||
assert!(!cb.record_success());
|
||||
}
|
||||
|
||||
/// `mark_startup_emitted` must fire exactly once so a fresh boot emits a
|
||||
/// single `SystemStartup` mark for `memory_tree_db` (C1).
|
||||
#[test]
|
||||
fn startup_mark_fires_exactly_once() {
|
||||
let cb = CircuitBreaker::new();
|
||||
assert!(cb.mark_startup_emitted());
|
||||
assert!(!cb.mark_startup_emitted());
|
||||
assert!(!cb.mark_startup_emitted());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,11 +1,74 @@
|
||||
//! Read and verify chunk and summary `.md` files from the content store.
|
||||
|
||||
use std::path::Path;
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
|
||||
use super::atomic::sha256_hex;
|
||||
use super::compose::split_front_matter;
|
||||
use crate::openhuman::memory::util::redact::redact;
|
||||
|
||||
/// Resolve a DB-stored relative forward-slash path against `content_root`,
|
||||
/// rejecting any traversal (`..`), absolute, or non-normal component.
|
||||
///
|
||||
/// The `raw_refs` / `content_path` values are treated as **untrusted** at the
|
||||
/// read boundary: although the write path slugifies/sanitizes them, a future
|
||||
/// ingest source or DB tamper could store `../../etc/passwd` and turn this
|
||||
/// reader into an arbitrary file-disclosure primitive that feeds the LLM
|
||||
/// context. We therefore (1) reject any `..`/absolute/prefix component before
|
||||
/// touching disk and (2) — when the target exists — canonicalize the resolved
|
||||
/// path and assert it stays under the canonicalized `content_root`.
|
||||
fn resolve_within_content_root(content_root: &Path, rel_path: &str) -> anyhow::Result<PathBuf> {
|
||||
// Reject absolute inputs outright. A leading `/` (or a Windows drive/UNC
|
||||
// prefix) would otherwise split into an empty leading component that gets
|
||||
// silently skipped, treating `/etc/passwd` as a relative path under the
|
||||
// content root rather than flagging the obvious traversal attempt.
|
||||
if Path::new(rel_path).is_absolute() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"[content_store::read] rejected absolute path in path_hash={}",
|
||||
redact(rel_path),
|
||||
));
|
||||
}
|
||||
|
||||
let mut abs = content_root.to_path_buf();
|
||||
for component in rel_path.split('/') {
|
||||
// Skip empty components from leading/double/trailing slashes.
|
||||
if component.is_empty() || component == "." {
|
||||
continue;
|
||||
}
|
||||
// Reject anything that is not a plain file/dir name: `..`, absolute
|
||||
// roots, Windows prefixes, etc.
|
||||
match Path::new(component).components().next() {
|
||||
Some(Component::Normal(_)) => abs.push(component),
|
||||
_ => {
|
||||
return Err(anyhow::anyhow!(
|
||||
"[content_store::read] rejected unsafe path component in path_hash={}",
|
||||
redact(rel_path),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Defense in depth: if the file exists, canonicalize and confirm
|
||||
// containment. (canonicalize requires the path to exist, so this is a
|
||||
// no-op for not-yet-created files — the component check above already
|
||||
// blocks traversal in that case.)
|
||||
if abs.exists() {
|
||||
let canon_root = content_root
|
||||
.canonicalize()
|
||||
.unwrap_or_else(|_| content_root.to_path_buf());
|
||||
let canon_abs = abs
|
||||
.canonicalize()
|
||||
.map_err(|e| anyhow::anyhow!("[content_store::read] canonicalize failed: {e}"))?;
|
||||
if !canon_abs.starts_with(&canon_root) {
|
||||
return Err(anyhow::anyhow!(
|
||||
"[content_store::read] resolved path escapes content_root for path_hash={}",
|
||||
redact(rel_path),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
Ok(abs)
|
||||
}
|
||||
|
||||
/// The result of reading a chunk file from disk.
|
||||
pub struct ChunkFileContents {
|
||||
/// The Markdown body (everything after the closing `---` of the front-matter).
|
||||
@@ -175,14 +238,9 @@ pub fn read_chunk_body(
|
||||
}
|
||||
|
||||
let content_root = config.memory_tree_content_root();
|
||||
// Reconstruct the absolute path from the stored relative forward-slash path.
|
||||
let abs_path = {
|
||||
let mut p = content_root.clone();
|
||||
for component in rel_path.split('/') {
|
||||
p.push(component);
|
||||
}
|
||||
p
|
||||
};
|
||||
// Reconstruct the absolute path from the stored relative forward-slash
|
||||
// path, rejecting any traversal and confirming containment.
|
||||
let abs_path = resolve_within_content_root(&content_root, &rel_path)?;
|
||||
|
||||
log::debug!(
|
||||
"[content_store::read] read_chunk_body chunk_id={} path_hash={}",
|
||||
@@ -237,10 +295,20 @@ fn read_chunk_body_from_raw(
|
||||
let content_root = config.memory_tree_content_root();
|
||||
let mut parts: Vec<String> = Vec::with_capacity(refs.len());
|
||||
for r in refs {
|
||||
let mut abs = content_root.clone();
|
||||
for component in r.path.split('/') {
|
||||
abs.push(component);
|
||||
}
|
||||
// Treat the DB-stored ref path as untrusted: reject traversal /
|
||||
// absolute paths and confirm the resolved path stays under
|
||||
// content_root before reading. Skip (don't fail the whole chunk) on a
|
||||
// rejected ref, matching the best-effort policy for per-file errors.
|
||||
let abs = match resolve_within_content_root(&content_root, &r.path) {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
log::warn!(
|
||||
"[content_store::read] raw_ref rejected path_hash={} err={e}",
|
||||
redact(&r.path)
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let bytes = match std::fs::read(&abs) {
|
||||
Ok(b) => b,
|
||||
Err(e) => {
|
||||
@@ -300,13 +368,7 @@ pub fn read_summary_body(
|
||||
let (rel_path, expected_sha256) = pointers;
|
||||
|
||||
let content_root = config.memory_tree_content_root();
|
||||
let abs_path = {
|
||||
let mut p = content_root.clone();
|
||||
for component in rel_path.split('/') {
|
||||
p.push(component);
|
||||
}
|
||||
p
|
||||
};
|
||||
let abs_path = resolve_within_content_root(&content_root, &rel_path)?;
|
||||
|
||||
log::debug!(
|
||||
"[content_store::read] read_summary_body summary_id={} path_hash={}",
|
||||
@@ -605,6 +667,55 @@ mod tests {
|
||||
assert_eq!(body, "bcd");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn read_chunk_body_from_raw_rejects_path_traversal() {
|
||||
use crate::openhuman::memory_store::chunks::store::RawRef;
|
||||
|
||||
let dir = TempDir::new().unwrap();
|
||||
let mut cfg = crate::openhuman::config::Config::default();
|
||||
cfg.workspace_dir = dir.path().to_path_buf();
|
||||
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
std::fs::create_dir_all(&content_root).unwrap();
|
||||
std::fs::write(content_root.join("safe.txt"), "safe").unwrap();
|
||||
|
||||
// A secret sitting next to content_root that a traversal ref tries to
|
||||
// reach. The traversal ref must be skipped, leaving only the safe body.
|
||||
let outside = content_root.parent().unwrap().join("secret.txt");
|
||||
std::fs::write(&outside, "TOP SECRET").unwrap();
|
||||
|
||||
let refs = vec![
|
||||
RawRef {
|
||||
path: "../secret.txt".into(),
|
||||
start: 0,
|
||||
end: None,
|
||||
},
|
||||
RawRef {
|
||||
path: "safe.txt".into(),
|
||||
start: 0,
|
||||
end: None,
|
||||
},
|
||||
];
|
||||
|
||||
let body = read_chunk_body_from_raw(&cfg, &refs).unwrap();
|
||||
assert_eq!(body, "safe");
|
||||
assert!(!body.contains("SECRET"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_within_content_root_rejects_traversal_and_absolute() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let root = dir.path();
|
||||
|
||||
assert!(resolve_within_content_root(root, "../escape.md").is_err());
|
||||
assert!(resolve_within_content_root(root, "a/../../escape.md").is_err());
|
||||
assert!(resolve_within_content_root(root, "/etc/passwd").is_err());
|
||||
|
||||
// Safe relative paths resolve correctly.
|
||||
let ok = resolve_within_content_root(root, "sub/dir/file.md").unwrap();
|
||||
assert_eq!(ok, root.join("sub").join("dir").join("file.md"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn read_chunk_body_roundtrips_from_staged_content_pointer() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
|
||||
@@ -49,8 +49,12 @@ impl Sandbox for BubblewrapSandbox {
|
||||
"/dev",
|
||||
"--proc",
|
||||
"/proc",
|
||||
"--bind",
|
||||
"/tmp",
|
||||
// Private, empty /tmp inside the sandbox instead of sharing the
|
||||
// host's read-write /tmp. Mirrors firejail.rs's `--private=home`
|
||||
// least-privilege posture: the agent gets a scratch tmp that is
|
||||
// discarded on exit and cannot read/overwrite other processes'
|
||||
// host temp files, lockfiles, sockets, or staged secrets.
|
||||
"--tmpfs",
|
||||
"/tmp",
|
||||
"--unshare-all",
|
||||
"--die-with-parent",
|
||||
@@ -180,4 +184,37 @@ mod tests {
|
||||
"must include /proc mount"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bubblewrap_wrap_command_uses_private_tmpfs_not_host_tmp() {
|
||||
let sandbox = BubblewrapSandbox;
|
||||
let mut cmd = Command::new("echo");
|
||||
sandbox.wrap_command(&mut cmd).unwrap();
|
||||
|
||||
let args: Vec<String> = cmd
|
||||
.get_args()
|
||||
.map(|s| s.to_string_lossy().to_string())
|
||||
.collect();
|
||||
|
||||
// The sandbox must mount a private tmpfs over /tmp ...
|
||||
let tmpfs_pos = args.iter().position(|a| a == "--tmpfs");
|
||||
assert!(
|
||||
tmpfs_pos.is_some(),
|
||||
"must include --tmpfs for a private /tmp"
|
||||
);
|
||||
assert_eq!(
|
||||
args.get(tmpfs_pos.unwrap() + 1).map(String::as_str),
|
||||
Some("/tmp"),
|
||||
"--tmpfs must target /tmp"
|
||||
);
|
||||
|
||||
// ... and must NOT read-write bind the host /tmp into the sandbox.
|
||||
let has_tmp_rw_bind = args
|
||||
.windows(3)
|
||||
.any(|w| w[0] == "--bind" && w[1] == "/tmp" && w[2] == "/tmp");
|
||||
assert!(
|
||||
!has_tmp_rw_bind,
|
||||
"must NOT bind host /tmp read-write into the sandbox"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
+29
-5
@@ -7909,10 +7909,19 @@ async fn public_paths_accessible_without_token() {
|
||||
);
|
||||
}
|
||||
|
||||
// Paths that bypass auth but return non-2xx for unrelated reasons
|
||||
// (missing required query params, no WebSocket upgrade headers, etc.).
|
||||
// The invariant is that the auth middleware does NOT reject them with 401.
|
||||
for path in ["/auth/telegram", "/events", "/ws/dictation"] {
|
||||
// Paths that bypass the *middleware* header check but return non-2xx for
|
||||
// unrelated reasons (missing required query params, no WebSocket upgrade
|
||||
// headers, etc.). The invariant here is that the auth middleware does NOT
|
||||
// reject them with 401.
|
||||
//
|
||||
// NOTE: `/events` and `/ws/dictation` enforce their own credential inside
|
||||
// the handler (bind token / bearer / `?token=`), so an unauthenticated GET
|
||||
// to those returns 401 from the HANDLER (not the middleware). `/events`
|
||||
// requires a `client_id` query param, so an empty request is rejected by
|
||||
// Axum's extractor (400) before the handler's own auth runs — it stays in
|
||||
// this `!= 401` group. `/ws/dictation` is asserted separately below now
|
||||
// that it is authenticated at the upgrade boundary (C4 / issue #1924).
|
||||
for path in ["/auth/telegram", "/events"] {
|
||||
let resp = client
|
||||
.get(format!("{base}{path}"))
|
||||
.send()
|
||||
@@ -7921,11 +7930,26 @@ async fn public_paths_accessible_without_token() {
|
||||
assert_ne!(
|
||||
resp.status(),
|
||||
StatusCode::UNAUTHORIZED,
|
||||
"public path {path} must not be auth-gated (got {})",
|
||||
"public path {path} must not be auth-gated by the middleware (got {})",
|
||||
resp.status()
|
||||
);
|
||||
}
|
||||
|
||||
// `/ws/dictation` is now authenticated at the upgrade boundary (C4 /
|
||||
// issue #1924): an unauthenticated request (no bearer header, no
|
||||
// `?token=`) must be rejected with 401 by the handler before any upgrade.
|
||||
let resp = client
|
||||
.get(format!("{base}/ws/dictation"))
|
||||
.send()
|
||||
.await
|
||||
.unwrap_or_else(|e| panic!("GET /ws/dictation: {e}"));
|
||||
assert_eq!(
|
||||
resp.status(),
|
||||
StatusCode::UNAUTHORIZED,
|
||||
"/ws/dictation must reject unauthenticated upgrades with 401 (got {})",
|
||||
resp.status()
|
||||
);
|
||||
|
||||
rpc_join.abort();
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user