Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

server: restore weak workspace web access

Browse Source
This commit is contained in:
Hare
2026-08-02 03:24:54 +09:00
parent e530150e43
commit e5f3c20f64
5 changed files with 34 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server.rs
+17 -75
View File
@@ -3786,27 +3786,9 @@ async fn scoped_list_runtimes(
async fn scoped_workspace_protocol_ws(
State(api): State<WorkspaceApi>,
AxumPath(workspace_id): AxumPath<String>,
headers: HeaderMap,
ws: axum::extract::ws::WebSocketUpgrade,
) -> std::result::Result<Response, Response> {
validate_workspace_scope(&api, &workspace_id).map_err(|error| error.into_response())?;
let actor = resolve_actor(&api, &headers)
.await
.map_err(|error| error.into_response())?
.ok_or_else(|| StatusCode::UNAUTHORIZED.into_response())?;
let workspace = api
.store
.get_workspace(&workspace_id)
.await
.map_err(|error| ApiError::from(error).into_response())?
.ok_or_else(|| StatusCode::NOT_FOUND.into_response())?;
if workspace
.owner_account_id
.as_deref()
.is_some_and(|owner| owner != actor.account_id.as_str())
{
return Err(StatusCode::FORBIDDEN.into_response());
}
Ok(ws
.on_upgrade(move |socket| {
crate::workspace_subscription::serve_workspace_subscription(api, socket)
@@ -5150,9 +5132,8 @@ async fn post_passkey_registration_complete(
)
})?;
let credential_id = passkey_credential_id(&passkey)?;
let registered_at = crate::auth::now_rfc3339();
api.store.upsert_passkey_and_claim_legacy_workspace_owner(
&PasskeyCredentialRecord {
api.store
.upsert_passkey_credential(&PasskeyCredentialRecord {
credential_id,
user_id: user.user_id.clone(),
public_key_cose: serde_json::to_string(&passkey).map_err(|error| {
@@ -5160,13 +5141,9 @@ async fn post_passkey_registration_complete(
})?,
transports_json: None,
sign_count: 0,
created_at: registered_at.clone(),
created_at: crate::auth::now_rfc3339(),
last_used_at: None,
},
api.workspace_id(),
&user.account_id,
&registered_at,
)?;
})?;
issue_browser_session_response(&api, user)
}
@@ -8881,7 +8858,6 @@ mod tests {
use std::{fs, sync::Arc};
use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tower::ServiceExt;
use worker_runtime::resource::BackendResourceClient;
use worker_runtime::working_directory::WorkingDirectoryMaterializer;
@@ -8891,9 +8867,8 @@ mod tests {
WorkerSpawnIntent,
};
use crate::store::{
AccountRecord, MemoryDocumentRecord, MemoryStagingRecord, ObjectiveRecord,
ObjectiveResourceRecord, ObjectiveTicketLinkRecord, SqliteWorkspaceStore, UserRecord,
WorkspaceRecord,
MemoryDocumentRecord, MemoryStagingRecord, ObjectiveRecord, ObjectiveResourceRecord,
ObjectiveTicketLinkRecord, SqliteWorkspaceStore, WorkspaceRecord,
};
const TEST_WORKSPACE_ID: &str = "0192f0e8-4d84-7d6e-a000-000000000001";
@@ -12791,7 +12766,7 @@ mod tests {
}
#[tokio::test]
async fn workspace_subscription_requires_browser_session() {
async fn workspace_subscription_uses_legacy_scoped_access_without_browser_session() {
let dir = tempfile::tempdir().unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
@@ -12799,51 +12774,20 @@ mod tests {
let server = tokio::spawn(async move {
let _ = axum::serve(listener, app).await;
});
let error = tokio_tungstenite::connect_async(format!(
let (mut socket, response) = tokio_tungstenite::connect_async(format!(
"ws://{address}/api/w/{TEST_WORKSPACE_ID}/protocol/ws"
))
.await
.unwrap_err();
let tokio_tungstenite::tungstenite::Error::Http(response) = error else {
panic!("expected HTTP authentication rejection");
};
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
.unwrap();
assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS);
socket.close(None).await.unwrap();
server.abort();
}
#[tokio::test]
async fn workspace_subscription_returns_authenticated_workspace_snapshot() {
async fn workspace_subscription_returns_workspace_snapshot() {
let dir = tempfile::tempdir().unwrap();
let api = test_api(dir.path()).await;
let account = AccountRecord {
account_id: "account-test".to_string(),
kind: "user".to_string(),
handle: "tester".to_string(),
display_name: "Tester".to_string(),
created_at: TEST_CREATED_AT.to_string(),
updated_at: TEST_CREATED_AT.to_string(),
};
let user = UserRecord {
user_id: "user-test".to_string(),
account_id: account.account_id.clone(),
handle: account.handle.clone(),
display_name: account.display_name.clone(),
created_at: TEST_CREATED_AT.to_string(),
updated_at: TEST_CREATED_AT.to_string(),
};
api.store.upsert_account(&account).unwrap();
api.store.upsert_user(&user).unwrap();
let session = issue_browser_session_response(&api, user).unwrap();
let cookie = session
.headers()
.get(SET_COOKIE)
.unwrap()
.to_str()
.unwrap()
.split(';')
.next()
.unwrap()
.to_string();
let spawn_request = WorkerSpawnRequest {
intent: WorkerSpawnIntent::WorkspaceCompanion,
@@ -12874,13 +12818,11 @@ mod tests {
let server = tokio::spawn(async move {
let _ = axum::serve(listener, app).await;
});
let mut request = format!("ws://{address}/api/w/{TEST_WORKSPACE_ID}/protocol/ws")
.into_client_request()
.unwrap();
request
.headers_mut()
.insert(axum::http::header::COOKIE, cookie.parse().unwrap());
let (mut socket, _) = connect_async(request).await.unwrap();
let (mut socket, _) = connect_async(format!(
"ws://{address}/api/w/{TEST_WORKSPACE_ID}/protocol/ws"
))
.await
.unwrap();
let frame = protocol::subscription::SubscriptionFrame::new(
protocol::subscription::SubscriptionFramePayload::Request(
protocol::subscription::SubscriptionRequest::SubscribeEvents {
crates/workspace-server/src/store.rs
-221
View File
@@ -504,16 +504,6 @@ pub trait ControlPlaneStore: Send + Sync {
fn get_user_by_handle(&self, handle: &str) -> Result<Option<UserRecord>>;
fn any_user(&self) -> Result<Option<UserRecord>>;
fn upsert_passkey_credential(&self, record: &PasskeyCredentialRecord) -> Result<()>;
/// Temporary migration bridge for legacy local databases that predate
/// Workspace ownership. The first verified passkey atomically claims an
/// ownerless Workspace; remove after supported databases have owners.
fn upsert_passkey_and_claim_legacy_workspace_owner(
&self,
record: &PasskeyCredentialRecord,
workspace_id: &str,
account_id: &str,
updated_at: &str,
) -> Result<bool>;
fn get_passkey_credential(
&self,
credential_id: &str,
@@ -1463,52 +1453,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn upsert_passkey_and_claim_legacy_workspace_owner(
&self,
record: &PasskeyCredentialRecord,
workspace_id: &str,
account_id: &str,
updated_at: &str,
) -> Result<bool> {
self.with_conn(|conn| {
let tx = conn.unchecked_transaction()?;
let verified_passkeys = tx.query_row(
"SELECT COUNT(*) FROM passkey_credentials",
[],
|row| row.get::<_, i64>(0),
)?;
tx.execute(
r#"INSERT INTO passkey_credentials (credential_id, user_id, public_key_cose, transports_json, sign_count, created_at, last_used_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
ON CONFLICT(credential_id) DO UPDATE SET
public_key_cose = excluded.public_key_cose,
transports_json = excluded.transports_json,
sign_count = excluded.sign_count,
last_used_at = excluded.last_used_at"#,
params![record.credential_id, record.user_id, record.public_key_cose, record.transports_json, record.sign_count, record.created_at, record.last_used_at],
)?;
let claimed = if verified_passkeys == 0 {
tx.execute(
r#"UPDATE workspaces
SET owner_account_id = ?2, updated_at = ?3
WHERE workspace_id = ?1
AND owner_account_id IS NULL
AND EXISTS (
SELECT 1 FROM users
WHERE user_id = ?4
AND account_id = ?2
AND state = 'active'
)"#,
params![workspace_id, account_id, updated_at, record.user_id],
)? == 1
} else {
false
};
tx.commit()?;
Ok(claimed)
})
}
fn get_passkey_credential(
&self,
credential_id: &str,
@@ -5305,171 +5249,6 @@ CREATE TABLE ticket_assignment_operations (
);
}
#[tokio::test]
async fn first_verified_passkey_claims_legacy_ownerless_workspace_once() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
let now = "2026-07-30T00:00:00Z".to_string();
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: "legacy-local".to_string(),
owner_account_id: None,
display_name: "Legacy local".to_string(),
state: "active".to_string(),
created_at: now.clone(),
updated_at: now.clone(),
})
.await
.unwrap();
for suffix in ["first", "second"] {
store
.upsert_account(&AccountRecord {
account_id: format!("account-{suffix}"),
kind: "user".to_string(),
handle: suffix.to_string(),
display_name: suffix.to_string(),
created_at: now.clone(),
updated_at: now.clone(),
})
.unwrap();
store
.upsert_user(&UserRecord {
user_id: format!("user-{suffix}"),
account_id: format!("account-{suffix}"),
handle: suffix.to_string(),
display_name: suffix.to_string(),
created_at: now.clone(),
updated_at: now.clone(),
})
.unwrap();
}
let first = PasskeyCredentialRecord {
credential_id: "credential-first".to_string(),
user_id: "user-first".to_string(),
public_key_cose: "public-key-first".to_string(),
transports_json: None,
sign_count: 0,
created_at: now.clone(),
last_used_at: None,
};
assert!(
store
.upsert_passkey_and_claim_legacy_workspace_owner(
&first,
"legacy-local",
"account-first",
&now,
)
.unwrap()
);
let second = PasskeyCredentialRecord {
credential_id: "credential-second".to_string(),
user_id: "user-second".to_string(),
public_key_cose: "public-key-second".to_string(),
transports_json: None,
sign_count: 0,
created_at: now.clone(),
last_used_at: None,
};
assert!(
!store
.upsert_passkey_and_claim_legacy_workspace_owner(
&second,
"legacy-local",
"account-second",
&now,
)
.unwrap()
);
assert_eq!(
store
.get_workspace("legacy-local")
.await
.unwrap()
.unwrap()
.owner_account_id
.as_deref(),
Some("account-first")
);
}
#[tokio::test]
async fn first_verified_passkey_preserves_existing_workspace_owner() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
let now = "2026-07-30T00:00:00Z".to_string();
store
.upsert_account(&AccountRecord {
account_id: "account-owner".to_string(),
kind: "user".to_string(),
handle: "owner".to_string(),
display_name: "Owner".to_string(),
created_at: now.clone(),
updated_at: now.clone(),
})
.unwrap();
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: "owned".to_string(),
owner_account_id: Some("account-owner".to_string()),
display_name: "Owned".to_string(),
state: "active".to_string(),
created_at: now.clone(),
updated_at: now.clone(),
})
.await
.unwrap();
store
.upsert_account(&AccountRecord {
account_id: "account-first".to_string(),
kind: "user".to_string(),
handle: "first".to_string(),
display_name: "First".to_string(),
created_at: now.clone(),
updated_at: now.clone(),
})
.unwrap();
store
.upsert_user(&UserRecord {
user_id: "user-first".to_string(),
account_id: "account-first".to_string(),
handle: "first".to_string(),
display_name: "First".to_string(),
created_at: now.clone(),
updated_at: now.clone(),
})
.unwrap();
assert!(
!store
.upsert_passkey_and_claim_legacy_workspace_owner(
&PasskeyCredentialRecord {
credential_id: "credential-first".to_string(),
user_id: "user-first".to_string(),
public_key_cose: "public-key-first".to_string(),
transports_json: None,
sign_count: 0,
created_at: now.clone(),
last_used_at: None,
},
"owned",
"account-first",
&now,
)
.unwrap()
);
assert_eq!(
store
.get_workspace("owned")
.await
.unwrap()
.unwrap()
.owner_account_id
.as_deref(),
Some("account-owner")
);
}
#[tokio::test]
async fn account_and_login_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
web/workspace/src/lib/workspace/console/worker-console.ui.test.ts
-8
View File
@@ -731,12 +731,4 @@ Deno.test("Workspace Worker list and Console share the multiplexed connection",
multiplexer.includes("frame: 'worker_protocol'"),
"Sidebar and Console should share one Workspace multiplexer and route Worker methods through a subscription lane",
);
assert(
multiplexer.includes("loadWhoami") &&
multiplexer.includes("actor === null") &&
multiplexer.includes("AUTH_RECHECK_DELAY_MS") &&
multiplexer.includes("Register or sign in on the Account page") &&
consolePage.includes("workspace_subscription_auth_required"),
"Workspace multiplexer should replace a tight unauthenticated WebSocket loop with an actionable auth status and bounded recheck",
);
});
web/workspace/src/lib/workspace/multiplexer.ts
+14 -43
View File
@@ -6,7 +6,6 @@ import type {
SubscriptionId,
} from '$lib/generated/protocol';
import { workspaceApiPath } from '$lib/workspace/api/http';
import { loadWhoami } from '$lib/workspace/auth/api';
type Listener = {
onFrame(frame: SubscriptionFrame): void;
@@ -27,9 +26,6 @@ export type WorkspaceMultiplexerSubscription = {
};
const multiplexers = new Map<string, WorkspaceMultiplexer>();
const RECONNECT_DELAY_MS = 500;
const AUTH_RECHECK_DELAY_MS = 5_000;
const AUTH_REQUIRED_MESSAGE = 'Authentication required. Register or sign in on the Account page.';
export function workspaceMultiplexer(workspaceId: string): WorkspaceMultiplexer {
let multiplexer = multiplexers.get(workspaceId);
@@ -93,45 +89,20 @@ export class WorkspaceMultiplexer {
});
socket.addEventListener('message', (event) => this.#receive(String(event.data)));
socket.addEventListener('error', () => socket.close());
socket.addEventListener('close', () => void this.#handleSocketClose(socket));
}
async #handleSocketClose(socket: WebSocket): Promise<void> {
if (this.#socket !== socket) return;
this.#socket = null;
this.#requests.clear();
this.#runtimeSubscriptions.clear();
for (const subscription of this.#subscriptions.values()) {
subscription.requestId = null;
subscription.subscriptionId = null;
}
if (this.#closed || this.#subscriptions.size === 0) return;
let authenticationRequired = false;
try {
authenticationRequired = (await loadWhoami()).actor === null;
} catch {
// A failed auth probe is treated as a transient network/backend failure.
}
if (this.#closed || this.#subscriptions.size === 0 || this.#socket) return;
const message = authenticationRequired
? AUTH_REQUIRED_MESSAGE
: 'Workspace subscription disconnected';
for (const subscription of this.#subscriptions.values()) {
subscription.listener.onStatus?.('closed', message);
}
this.#scheduleReconnect(
authenticationRequired ? AUTH_RECHECK_DELAY_MS : RECONNECT_DELAY_MS,
);
}
#scheduleReconnect(delayMs: number): void {
if (this.#reconnectTimer) clearTimeout(this.#reconnectTimer);
this.#reconnectTimer = setTimeout(() => {
this.#reconnectTimer = null;
this.#ensureConnected();
}, delayMs);
socket.addEventListener('close', () => {
if (this.#socket !== socket) return;
this.#socket = null;
this.#requests.clear();
this.#runtimeSubscriptions.clear();
for (const subscription of this.#subscriptions.values()) {
subscription.requestId = null;
subscription.subscriptionId = null;
subscription.listener.onStatus?.('closed', 'Workspace subscription disconnected');
}
if (!this.#closed && this.#subscriptions.size > 0) {
this.#reconnectTimer = setTimeout(() => this.#ensureConnected(), 500);
}
});
}
#sendSubscribe(subscription: ActiveSubscription): void {
web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte
+3 -21
View File
@@ -564,29 +564,11 @@
];
}
},
onStatus: (status, message) => {
onStatus: (status) => {
if (token !== reloadToken) return;
const authenticationRequired = message?.startsWith("Authentication required") ?? false;
protocolState = authenticationRequired
? "error"
: status === "open"
? "connecting"
: status;
if (authenticationRequired && message) {
streamDiagnostics = [
...streamDiagnostics.filter(
(diagnostic) =>
diagnostic.code !== "workspace_subscription_auth_required",
),
{
code: "workspace_subscription_auth_required",
severity: "error",
message,
},
];
}
protocolState = status === "open" ? "connecting" : status;
if (status === "closed") {
rejectPendingCompletion(new Error(message ?? "Worker protocol WebSocket closed."));
rejectPendingCompletion(new Error("Worker protocol WebSocket closed."));
}
},
},