From 4fe7f02ebb2421a45d266d4bb7c272cc24a22e10 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Mon, 3 Aug 2026 00:18:39 +0200 Subject: [PATCH] version 0.4.0 --- Cargo.lock | 116 +-- Cargo.toml | 2 +- README.md | 2 + UPSTREAM_SYNC.md | 1 + crates/proton-drive-rs/Cargo.toml | 2 +- crates/proton-drive-rs/src/client.rs | 156 +--- crates/proton-drive-rs/src/crypto.rs | 29 +- crates/proton-drive-rs/src/lib.rs | 4 +- crates/proton-drive-rs/src/public_link.rs | 799 ++++++++++++++++--- crates/proton-drive-rs/src/revision.rs | 24 +- crates/proton-drive-rs/src/transport.rs | 492 ++++++++++++ crates/proton-drive-rs/tests/live_sharing.rs | 318 +++++++- crates/proton-drive-rs/tests/spawnable.rs | 48 +- crates/proton-sdk/src/http.rs | 136 +++- 14 files changed, 1838 insertions(+), 291 deletions(-) create mode 100644 crates/proton-drive-rs/src/transport.rs diff --git a/Cargo.lock b/Cargo.lock index 6690128..a2f0ac2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -42,9 +42,9 @@ dependencies = [ [[package]] name = "aes" -version = "0.9.1" +version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1fc76eaeac4c9164506c466d4ffdd8ec9d0c5bf57ee97177c4d8eceb3a0e138" +checksum = "f8eb277bec05f56a0e0591f155a484cbd0f4f07ff2905051a48c72f004f7ed58" dependencies = [ "cipher 0.5.2", "cpubits", @@ -72,7 +72,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fdf011db2e21ce0d575593d749db5554b47fed37aff429e4dc50bc91ac93a028" dependencies = [ "aead 0.6.1", - "aes 0.9.1", + "aes 0.9.2", "cipher 0.5.2", "ctr 0.10.1", "ghash 0.6.0", @@ -118,9 +118,9 @@ dependencies = [ [[package]] name = "async-compression" -version = "0.4.42" +version = "0.4.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e79b3f8a79cccc2898f31920fc69f304859b3bd567490f75ebf51ae1c792a9ac" +checksum = "3976abdc8fe7d1133d43d304afd42abdf5bc3e1319d263d223bde07b5efc4be8" dependencies = [ "compression-codecs", "compression-core", @@ -136,7 +136,7 @@ checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec" dependencies = [ "proc-macro2", "quote", - "syn 3.0.0", + "syn 3.0.3", ] [[package]] @@ -186,6 +186,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b25655df2c3cdd83c5e5b293b88acd880332b2ddadd7c30ac43144fdc0033da9" + [[package]] name = "base64ct" version = "1.8.3" @@ -194,11 +200,11 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" [[package]] name = "bcrypt" -version = "0.19.2" +version = "0.19.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f3c067aa24dd4ed5c79cf222a38f260c8f23d3b82a062fba3f28c6fe563b753" +checksum = "a0cd0bd35a28836d528d2b58ad499bc3c5641d59379421b1be9eeb0c2f2b912a" dependencies = [ - "base64", + "base64 0.23.0", "blowfish 0.10.0", "getrandom 0.4.3", "subtle", @@ -357,9 +363,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.3.0" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8" +checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9" dependencies = [ "find-msvc-tools", "jobserver", @@ -817,13 +823,13 @@ dependencies = [ [[package]] name = "displaydoc" -version = "0.2.6" +version = "0.2.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] @@ -1215,9 +1221,9 @@ dependencies = [ [[package]] name = "http" -version = "1.4.2" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" dependencies = [ "bytes", "itoa", @@ -1254,18 +1260,18 @@ checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" [[package]] name = "hybrid-array" -version = "0.4.13" +version = "0.4.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c" +checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" dependencies = [ "typenum", ] [[package]] name = "hyper" -version = "1.10.1" +version = "1.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" +checksum = "d22053281f852e11534f5198498373cbb59295120a20771d90f7ed1897490a72" dependencies = [ "atomic-waker", "bytes", @@ -1302,7 +1308,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-util", @@ -1596,9 +1602,9 @@ checksum = "34b357333733e8260735ba5894eb928c02ecc69c78715f01a8019e7fa7f2db4c" [[package]] name = "libc" -version = "0.2.186" +version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "libm" @@ -1877,7 +1883,7 @@ dependencies = [ "aes-gcm 0.10.3", "aes-kw", "argon2", - "base64", + "base64 0.22.1", "bitfields", "block-padding", "blowfish 0.9.1", @@ -2030,10 +2036,10 @@ dependencies = [ [[package]] name = "proton-drive-rs" -version = "0.3.2" +version = "0.4.0" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "futures", "getrandom 0.4.3", @@ -2055,11 +2061,11 @@ dependencies = [ [[package]] name = "proton-sdk" -version = "0.3.2" +version = "0.4.0" dependencies = [ "aes-gcm 0.11.0", "async-trait", - "base64", + "base64 0.22.1", "bcrypt", "bytes", "chrono", @@ -2230,7 +2236,7 @@ version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04e9018c9d814e5f30cc16a0f03271aeab3571e609612d9fe78c1aa8d11c2f62" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-core", "futures-util", @@ -2336,9 +2342,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.42" +version = "0.23.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" dependencies = [ "aws-lc-rs", "once_cell", @@ -2362,9 +2368,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.15.0" +version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ "web-time", "zeroize", @@ -2504,14 +2510,14 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.0", + "syn 3.0.3", ] [[package]] name = "serde_json" -version = "1.0.150" +version = "1.0.151" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" dependencies = [ "itoa", "memchr", @@ -2528,7 +2534,7 @@ checksum = "8d3b1629de253c70a0508c3899572da79ca359fdab27c7920ff00406df418906" dependencies = [ "proc-macro2", "quote", - "syn 3.0.0", + "syn 3.0.3", ] [[package]] @@ -2652,18 +2658,18 @@ checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" [[package]] name = "snafu" -version = "0.9.1" +version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d1a012328be2e3f5d5f6f3218147ca02588cea4cb865e876849ab6debcf36522" +checksum = "e45cb604038abb7b926b679887b3226d8d0f23874b66623625a0454be425a4b7" dependencies = [ "snafu-derive", ] [[package]] name = "snafu-derive" -version = "0.9.1" +version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f103c50866b8743da9429b8a581d81a27c2d3a9c4ac7df8f8571c1dd7896eda" +checksum = "287f59010008f0d7cf5e3b03196d666c1acc46c8d3e9cf34c28a1a7157601e72" dependencies = [ "heck", "proc-macro2", @@ -2728,9 +2734,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.0" +version = "3.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2fac314a64dc9a36e61a9eb4261a5e9bbfbc922b27e518af97bc32b926cf967" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" dependencies = [ "proc-macro2", "quote", @@ -2800,7 +2806,7 @@ checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" dependencies = [ "proc-macro2", "quote", - "syn 3.0.0", + "syn 3.0.3", ] [[package]] @@ -2830,9 +2836,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.53.0" +version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d988bcd52dbe076d3d46903332f58c912b87a2c49b1428419a5845154762ffee" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ "bytes", "libc", @@ -2845,13 +2851,13 @@ dependencies = [ [[package]] name = "tokio-macros" -version = "2.7.1" +version = "2.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] @@ -2866,9 +2872,9 @@ dependencies = [ [[package]] name = "tokio-util" -version = "0.7.18" +version = "0.7.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" dependencies = [ "bytes", "futures-core", @@ -3452,18 +3458,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.54" +version = "0.8.55" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7cbbc0a705a0fd05cc3676525980d2bf5a9bc4adac6d6475209a7887cf59d19" +checksum = "b5a105cd7b140f6eeec8acff2ea38135d3cab283ada58540f629fe51e46696eb" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.54" +version = "0.8.55" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e2e817b7b52d0c7358d3246da9d69935ebb18116b2b102b4230dac079b4862f5" +checksum = "0fe976fb70c78cd64cccfe3a6fc142244e8a77b70959b30faf9d0ac37ee228eb" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 9821f2b..defeb14 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = ["crates/proton-sdk", "crates/proton-drive-rs"] [workspace.package] -version = "0.3.2" +version = "0.4.0" edition = "2024" license = "MIT" repository = "https://github.com/narrrl/proton-sdk-rs" diff --git a/README.md b/README.md index 21882af..8031c00 100644 --- a/README.md +++ b/README.md @@ -270,6 +270,8 @@ Moving a file or folder within the same volume is an offline cryptographic opera | | External Invites | ✅ | Invite non-Proton users (`invite_external_users`) and track registration status. | | | Bookmarks | ✅ | Save and list public link bookmarks (`create_bookmark`, `list_bookmarks`). | | | Incoming Invites | ✅ | List, accept, or reject invitations shared *with* the current user. | +| | Public Link (consume) | ✅ | Open someone else's link with no Proton account (`ProtonDrivePublicLinkClient`): SRP handshake, custom passwords, browse and download the shared subtree. | +| | Public Link streaming | ✅ | Seekable range reads over a shared file (`open_revision` → `RevisionReader::read_at`), plus thumbnails and in-place session renewal — a visitor can stream a large file without downloading it. | | **Photos** | Photos Timeline | ✅ | `ProtonPhotosClient` maps photostream, timeline enumeration, and photo downloads. | | | Photo Uploads | ✅ | Uploading photos with `PhotoUploadMetadata` (capture time, tags, grouping). | | | Albums (read) | ✅ | List albums (`enumerate_album_node_uids`) and their photos (`enumerate_album`); album nodes carry `Node::album`. | diff --git a/UPSTREAM_SYNC.md b/UPSTREAM_SYNC.md index dc04f4e..0847844 100644 --- a/UPSTREAM_SYNC.md +++ b/UPSTREAM_SYNC.md @@ -32,3 +32,4 @@ gitignored, so this file is the only durable record of what we last reviewed. | 2026-07-24 | `1b6ff835` → `e0ad37e3` | `d2ffde20`, `724b6771` | both features were already ported from the TypeScript SDK, so this closed the remaining parity deltas. ported: **shared-by-me on the photos volume** (`SharingOperations.EnumerateSharedNodeUidsAsync` now takes a volume resolver — `page_shared_by_me(volume_id)` + `ProtonPhotosClient::enumerate_shared_node_uids`, empty when the account has no photos volume) and `SharedByMeLinkDto.ContextShareID` (kept `Option` — recent upstream addition, older backends omit it); **per-node move outcomes** (`move_nodes` now returns `Vec<(NodeUid, Result<()>)>` in input order, mirroring the breaking `MoveNodesAsync` result map; per-link envelope codes via the new `aggregate_outcomes`, which `check_aggregate` now sits on). cosmetic, not ported: `AnchorID` retyped `string` → `LinkId` (ours is `String`). diverged deliberately: we keep the batch endpoint `PUT volumes/{vid}/links/move-multiple` — upstream still loops per node behind a `// FIXME: … use the batch move endpoint`; and the drive-side listing keeps the `_by_me` suffix (upstream renamed both to `EnumerateSharedNodeUids`) because `proton-drive-linux` already consumes that name. | | 2026-07-22 | `2219a42f` → `1b6ff835` | `ded1b716`, `071f976f` | ported: **atomic small uploads** for buffered new files and revisions (`DriveSmallFileUpload` is an explicit opt-in because this crate has no remote feature-flag provider), and **photo duplicate detection** using name/content HMAC matching. already covered/no-op: `ee0b3fba` (Rust rename accepts `Option<&str>`), `193c26f1` (Rust in-memory cache already has bounded LRU eviction), `29d359e0` (digest representation refactor retains hex JSON), `5c370289` (C# disposal lifetime only), `95752055` (Photos cache construction only). deliberately not ported: `2bf49aae` / `2590d227` cache storage-format changes; Rust exposes a distinct string repository contract and ships no persistent backend to migrate/reset. | | 2026-07-31 | `e0ad37e3` → `941aafa5` | `42526703`, `f2fa871b`, `814ae781`, `c528a376` | ported: **node-name validation** (`validate_node_name`: empty or >255 chars rejected before create/rename; counted in `char`s — C# counts UTF-16 units, and neither is the server's own accounting); **album read ops** (`AlbumDto` + `enumerate_album_node_uids` / `enumerate_album`, `GET photos/volumes/{vid}/albums` and `…/albums/{lid}/children?Sort=Captured&Desc=1`, both paged on `AnchorID`; `LinkDetailsDto::folder_properties` reads an album's hash key out of the `Album` block, which also makes `parent_hash_key` work for album children); **photo + album node metadata** (`Node::photo` / `Node::album` — capture time, content hash, main/related photos, tags, album inclusions; photo count, cover photo, last activity — added as optional `Node` fields rather than `NodeKind` variants so persisted node blobs keep deserializing, mirroring how `membership` landed); **photo tag updates** (`update_photos`, `POST`/`DELETE photos/volumes/{vid}/links/{lid}/tags` via the new `ApiHttpClient::delete_with_body`, `POST …/favorite` for photos on our own photos volume, one outcome per input update like `move_nodes`); **shared-with-me photos/albums** (`ShareTargetType::is_photos_item` over `v2/sharedwithme` + the dedicated `GET photos/albums/shared-with-me`, whose rows carry their own `VolumeID`). deferred: `03b1cb7f` **favoriting shared photos** — needs `PhotosCrypto` + `PhotoTransferPayloadBuilder` (name/passphrase session keys rewrapped to the timeline root, related-photo fan-out, batches of 20); our `rewrap_message_to` / `recover_message_session_key` make it feasible, but until it lands `update_photos` fails that one update explicitly instead of silently no-oping. not ported: `4c1dd783` (`IsSharedPublicly` → `IsSharedByUrl` rename — `proton-drive-linux` consumes `is_shared_publicly`), `1477ea45` (C# user-facing error strings; our error text is our own), the second half of `42526703` (server-supplied names validated into a per-node `InvalidNameError` — we have no node error list to put it in). no-op: `82d605b7` (C# `FifoFlexibleSemaphore` cancel race — tokio's `Semaphore` is cancel-safe and drops queued waiters), `8bdd9420` (throw on resume-after-dispose; Rust ownership makes it unreachable), `ca11470f` + `c947fce7` (CExports/interop only). Album *writes* (create, add/remove photos) remain unported — upstream has no C# public API for them either. Live validation pending: every photos endpoint here needs a photos-volume account. | +| 2026-08-01 | — (no upstream delta) | — | **not an upstream port.** Extended the TypeScript-sourced public-link consume side with a streaming surface that has **no TS or C# counterpart** and is ours: `open_revision` / `open_revision_at` / `download_range` / `download_thumbnail` / `enumerate_thumbnails` / `refresh_session` / `with_max_inflight_blocks` on `ProtonDrivePublicLinkClient`. Enabling change: `RevisionReader` was welded to `ProtonDriveClient`, so it now takes a `RevisionTransport` (new `transport.rs` — block permits + a route-prefixed `ApiHttpClient` + a `BlockSession` for renewal); `fetch_revision_blocks` and the block-size ranking moved there out of `client.rs`. Also `ApiHttpClient::without_token_refresh` (proton-sdk), because the anonymous session's refresh token is empty and the built-in 401 refresh masked the real 401. Perf on the visitor path: LRU node-key cache + iterative ancestor walk behind `SingleFlight` (was re-deriving the whole chain, at an S2K each, per child), `enumerate_nodes` chunked at 150 and fanned out (it did not chunk at all), `download_file_to` paged (a single un-paged GET truncated anything past 50 blocks) and concurrent. Live-verified against the test account, including the one uncertain fact the design rested on: the `drive/unauth/` revision response **does** carry `XAttr`, so a visitor resolves true per-block sizes and can seek. | diff --git a/crates/proton-drive-rs/Cargo.toml b/crates/proton-drive-rs/Cargo.toml index 49f40af..bedb734 100644 --- a/crates/proton-drive-rs/Cargo.toml +++ b/crates/proton-drive-rs/Cargo.toml @@ -14,7 +14,7 @@ readme = "../../README.md" documentation = "https://docs.rs/proton-drive-rs" [dependencies] -proton-sdk = { path = "../proton-sdk", version = "0.3.2" } +proton-sdk = { path = "../proton-sdk", version = "0.4.0" } tokio.workspace = true reqwest.workspace = true serde.workspace = true diff --git a/crates/proton-drive-rs/src/client.rs b/crates/proton-drive-rs/src/client.rs index adb0f88..fe9fb52 100644 --- a/crates/proton-drive-rs/src/client.rs +++ b/crates/proton-drive-rs/src/client.rs @@ -63,13 +63,13 @@ use crate::dtos::{ NodeNameAvailabilityRequest, NodeNameAvailabilityResponse, PhotoTagsRequest, PhotosAttributesDto, RenameLinkRequest, RevisionConflict, RevisionCreationRequest, RevisionCreationResponse, RevisionDto, RevisionListItemDto, RevisionListResponse, - RevisionMetadataResponse, RevisionResponse, RevisionUpdateRequest, ShareInvitationDto, - ShareInvitationsResponse, ShareMembersResponse, ShareMembershipSummaryDto, ShareResponse, - ShareTargetType, ShareUrlDto, ShareUrlsResponse, SharedAlbumsResponse, SharedByMeResponse, - SharedWithMeResponse, SmallFileUploadMetadataRequest, SmallRevisionUploadMetadataRequest, - SmallUploadResponse, ThumbnailBlockListRequest, ThumbnailBlockListResponse, - ThumbnailCreationRequest, ThumbnailDto, TimelinePhotoListResponse, UpdatePermissionsRequest, - VolumeCreationRequest, VolumeEventDto, VolumeEventListResponse, VolumeTrashResponse, + RevisionMetadataResponse, RevisionUpdateRequest, ShareInvitationDto, ShareInvitationsResponse, + ShareMembersResponse, ShareMembershipSummaryDto, ShareResponse, ShareTargetType, ShareUrlDto, + ShareUrlsResponse, SharedAlbumsResponse, SharedByMeResponse, SharedWithMeResponse, + SmallFileUploadMetadataRequest, SmallRevisionUploadMetadataRequest, SmallUploadResponse, + ThumbnailBlockListRequest, ThumbnailBlockListResponse, ThumbnailCreationRequest, ThumbnailDto, + TimelinePhotoListResponse, UpdatePermissionsRequest, VolumeCreationRequest, VolumeEventDto, + VolumeEventListResponse, VolumeTrashResponse, }; use crate::events::{DriveEvent, DriveEventScopeId}; use crate::node::{ @@ -87,19 +87,21 @@ use crate::sharing::{ PublicLink, ShareInvitation, ShareMember, ShareMembership, SharedWithMeItem, }; use crate::single_flight::SingleFlight; - -/// Content blocks are 4 MiB of plaintext each (C# `RevisionWriter.DefaultBlockSize`). -const DEFAULT_BLOCK_SIZE: usize = 1 << 22; +use crate::transport::{DEFAULT_BLOCK_SIZE, RevisionTransport, rank_block_sizes}; /// Maximum encrypted multipart size accepted by the atomic upload endpoint. const SMALL_UPLOAD_SIZE_LIMIT: usize = 128 * 1024; /// Maximum links per batch trash/restore/delete request (C# /// `NodeOperations.MaximumBatchCount`). -const MAX_BATCH_COUNT: usize = 150; +pub(crate) const MAX_BATCH_COUNT: usize = 150; + +/// Thumbnail block ids resolved per `POST volumes/{vid}/thumbnails` request +/// (C# `FileOperations.MaxThumbnailIdsPerRequest`). +pub(crate) const MAX_THUMBNAIL_IDS_PER_REQUEST: usize = 30; /// Link-detail batches fetched at once, per volume, when enumerating nodes. -const MAX_CONCURRENT_DETAIL_FETCHES: usize = 4; +pub(crate) const MAX_CONCURRENT_DETAIL_FETCHES: usize = 4; /// Nodes decrypted at once within one batch of link details. /// @@ -121,7 +123,7 @@ const MAX_CONCURRENT_NODE_BUILDS: usize = 8; /// Sized a little above one file's pipeline (an upload holds up to /// `MAX_BUFFERED_UPLOAD_BLOCKS + MAX_CONCURRENT_BLOCK_UPLOADS + 1` permits) so a /// lone transfer still saturates it and a second one is not starved. -const DEFAULT_MAX_INFLIGHT_BLOCKS: usize = 16; +pub(crate) const DEFAULT_MAX_INFLIGHT_BLOCKS: usize = 16; /// Content blocks uploaded concurrently within a single file. /// @@ -515,13 +517,6 @@ impl ProtonDriveClient { } /// The account client backing this Drive client's key chain. - /// The underlying HTTP client, for sibling modules that issue their own - /// requests against the same session (e.g. [`RevisionReader`] fetching - /// block bodies from storage). - pub(crate) fn http(&self) -> &ApiHttpClient { - &self.http - } - pub fn account(&self) -> &AccountClient { &self.account } @@ -587,6 +582,16 @@ impl ProtonDriveClient { self.block_slots.clone() } + /// What a [`RevisionReader`] needs from this client, and nothing else. + /// + /// The session is `Static` because an ordinary bearer session refreshes its + /// own tokens inside [`ApiHttpClient`] on a 401; by the time an error + /// reaches the transport there is nothing left to retry. The visitor path + /// supplies a session that can genuinely be renewed. + pub(crate) fn revision_transport(&self) -> RevisionTransport { + RevisionTransport::authenticated(self.http.clone(), self.block_slots()) + } + /// Resolve (and cache) the user's "My Files" root folder. pub async fn get_my_files_folder(&self) -> Result { self.ensure_my_files().await?; @@ -1361,7 +1366,7 @@ impl ProtonDriveClient { timer.success(); Ok(RevisionReader::new( - self.clone(), + self.revision_transport(), uid.clone(), revision_id, content_key, @@ -1639,26 +1644,11 @@ impl ProtonDriveClient { /// Plaintext size of each content block, in block order. /// - /// These sizes are not merely descriptive: [`RevisionReader`] accumulates - /// them to derive where each block *starts* in the plaintext. A vector that - /// overstates any block but the last therefore shifts every block after it, - /// and the reader splices those blocks at the wrong offsets — returning - /// full-length reads of the wrong bytes, with no error to notice. See - /// `revision.rs::padded_block_sizes_shift_later_blocks`, which pins that - /// consequence. - /// - /// So the sources are ranked by how much they can be trusted, and the - /// function refuses rather than guesses when none of them holds: - /// - /// 1. `Common.BlockSizes` from the revision's extended attributes — the - /// authoritative value, written by the uploading client. - /// 2. `Common.Size` — the total, from which all-but-last full blocks and a - /// final short block follow by subtraction. Sound for the standard - /// layout the uploader produces. - /// 3. Nothing usable. A single-block file cannot be misaligned (there is no - /// later block to shift), so it is assumed full-size and only its - /// reported length can be wrong. A multi-block file is an error: serving - /// it would mean serving misplaced bytes. + /// Decrypts the revision's extended attributes — *verified*, which is what + /// needs an account client and is therefore why this lives here rather than + /// in `transport` — and hands them to + /// [`rank_block_sizes`](crate::transport::rank_block_sizes), which owns the + /// ranking rules and the refusal. async fn resolve_block_sizes( &self, node_key: &PrivateKey, @@ -1687,40 +1677,7 @@ impl ProtonDriveClient { None => None, }; - let block = DEFAULT_BLOCK_SIZE as u64; - - if let Some(sizes) = common.as_ref().and_then(|c| c.block_sizes.as_ref()) - && sizes.len() == block_count - { - // A non-positive entry is as corrupting as a padded one — a - // zero-length block in the middle shifts everything after it — so a - // malformed vector is rejected outright rather than clamped. - if let Some(bad) = sizes.iter().position(|&n| n <= 0) { - return Err(ProtonError::invalid_operation(format!( - "revision {} records a non-positive size for block {bad}", - revision.id - ))); - } - return Ok(sizes.iter().map(|&n| n as u64).collect()); - } - - if let Some(total) = common.and_then(|c| c.size).filter(|&n| n >= 0) { - let total = total as u64; - return Ok((0..block_count) - .map(|i| total.saturating_sub(block * i as u64).min(block)) - .collect()); - } - - if block_count <= 1 { - return Ok(vec![block; block_count]); - } - - Err(ProtonError::invalid_operation(format!( - "revision {} has {block_count} blocks but no usable size information \ - (no BlockSizes, no Size); refusing to guess, since a wrong block \ - layout would serve misaligned content", - revision.id - ))) + rank_block_sizes(common.as_ref(), &revision.id, block_count) } /// Download and decrypt a file's thumbnail of the given type, if it has one. @@ -1814,8 +1771,6 @@ impl ProtonDriveClient { thumbnail_type: ThumbnailType, for_photos: bool, ) -> Result> { - const MAX_THUMBNAIL_IDS_PER_REQUEST: usize = 30; - let mut results: Vec = Vec::new(); // Group link ids by volume, preserving first-seen volume order. @@ -5087,50 +5042,21 @@ impl ProtonDriveClient { Ok(()) } + /// The revision's metadata and its full, ordered block table. + /// + /// The paging, ordering and contiguity rules are identical on the visitor + /// path, so they live in + /// [`RevisionTransport::list_blocks`](crate::transport::RevisionTransport) + /// and this is the authenticated entry point to them. pub(crate) async fn fetch_revision_blocks( &self, volume_id: &VolumeId, link_id: &LinkId, revision_id: &str, ) -> Result<(RevisionDto, Vec)> { - const PAGE_SIZE: i32 = 50; - - let mut blocks: Vec = Vec::new(); - let mut metadata: Option = None; - let mut from_index: i32 = 1; - - loop { - let path = format!( - "v2/volumes/{volume_id}/files/{link_id}/revisions/{revision_id}?FromBlockIndex={from_index}&PageSize={PAGE_SIZE}&NoBlockUrls=0" - ); - let response: RevisionResponse = self.http.get(&path).await?; - let mut revision = response.revision; - let page = std::mem::take(&mut revision.blocks); - let page_len = page.len(); - - if metadata.is_none() { - metadata = Some(revision); - } - blocks.extend(page); - - if page_len < PAGE_SIZE as usize { - break; - } - from_index = blocks.iter().map(|b| b.index).max().unwrap_or(from_index) + 1; - } - - blocks.sort_by_key(|b| b.index); - for (offset, block) in blocks.iter().enumerate() { - if block.index != offset as i32 + 1 { - return Err(ProtonError::invalid_operation( - "file contents are incomplete (non-contiguous blocks)", - )); - } - } - - let metadata = metadata - .ok_or_else(|| ProtonError::invalid_operation("revision returned no metadata"))?; - Ok((metadata, blocks)) + self.revision_transport() + .list_blocks(volume_id, link_id, revision_id) + .await } async fn ensure_my_files(&self) -> Result<()> { diff --git a/crates/proton-drive-rs/src/crypto.rs b/crates/proton-drive-rs/src/crypto.rs index f93c763..5cd1edc 100644 --- a/crates/proton-drive-rs/src/crypto.rs +++ b/crates/proton-drive-rs/src/crypto.rs @@ -249,19 +249,34 @@ pub async fn decrypt_extended_attributes_verified( Ok((attrs, status)) } +/// Decrypt a revision's extended attributes without verifying their signature. +/// +/// For callers that have the node key but no [`AccountClient`] to resolve the +/// author's public key through — namely an anonymous public-link visitor, who +/// has no address book and cannot look up a signer at all. +/// +/// Skipping verification loses nothing a visitor could otherwise have used: +/// [`decrypt_extended_attributes_verified`] returns the status as *metadata* and +/// callers do not gate on it (see the SDK's rule that a failed verification is +/// never an error). Decryption still proves the payload was encrypted to this +/// node key. +pub(crate) fn decrypt_extended_attributes_unverified( + node_key: &PrivateKey, + armored_xattr: &str, +) -> Result { + let json = node_key.decrypt_armored_message(armored_xattr)?; + serde_json::from_slice(&json).map_err(|e| { + ProtonError::invalid_operation(format!("deserialize extended attributes: {e}")) + }) +} + #[cfg(test)] mod tests { use super::*; use proton_sdk::crypto::generate_node_key; - /// Decrypt + deserialize XAttr without verification (the production path - /// uses [`decrypt_extended_attributes_verified`], which needs an account to - /// resolve author keys). fn decrypt_xattr(node_key: &PrivateKey, armored: &str) -> DecryptedExtendedAttributes { - let json = node_key - .decrypt_armored_message(armored) - .expect("decrypt xattr"); - serde_json::from_slice(&json).expect("deserialize xattr") + decrypt_extended_attributes_unverified(node_key, armored).expect("decrypt xattr") } #[test] diff --git a/crates/proton-drive-rs/src/lib.rs b/crates/proton-drive-rs/src/lib.rs index acc7b1c..10646ce 100644 --- a/crates/proton-drive-rs/src/lib.rs +++ b/crates/proton-drive-rs/src/lib.rs @@ -45,6 +45,7 @@ mod public_link; mod revision; mod sharing; mod single_flight; +mod transport; pub use cache::{CachedNodeInfo, DriveEntityCache}; pub use client::ProtonDriveClient; @@ -55,7 +56,8 @@ pub use event_manager::{ }; pub use events::{DriveEvent, DriveEventScopeId}; pub use node::{ - AlbumProperties, Node, NodeKind, PhotoProperties, RevisionState, Thumbnail, ThumbnailType, + AlbumProperties, FileThumbnail, Node, NodeKind, PhotoProperties, RevisionState, Thumbnail, + ThumbnailType, }; pub use photos::{ AlbumItem, PhotoTag, PhotoTagsUpdate, PhotoUploadMetadata, PhotosTimelineItem, diff --git a/crates/proton-drive-rs/src/public_link.rs b/crates/proton-drive-rs/src/public_link.rs index 09079c1..9569b0c 100644 --- a/crates/proton-drive-rs/src/public_link.rs +++ b/crates/proton-drive-rs/src/public_link.rs @@ -29,9 +29,15 @@ //! claimed, rather than a verification that silently passed. use std::collections::HashMap; +use std::num::NonZeroUsize; +use std::sync::Arc; +use async_trait::async_trait; use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64; +use futures::stream::{self, StreamExt, TryStreamExt}; +use lru::LruCache; +use tokio::sync::Semaphore; use proton_sdk::config::ProtonClientConfiguration; use proton_sdk::crypto::{ @@ -42,15 +48,21 @@ use proton_sdk::error::{ProtonError, Result}; use proton_sdk::http::{ApiHttpClient, Tokens, get_unauthenticated}; use proton_sdk::ids::{LinkId, NodeUid, SessionId, VolumeId}; -use crate::client::parse_public_link_url; -use crate::crypto::decrypt_link; +use crate::client::{ + DEFAULT_MAX_INFLIGHT_BLOCKS, MAX_BATCH_COUNT, MAX_CONCURRENT_DETAIL_FETCHES, + MAX_THUMBNAIL_IDS_PER_REQUEST, parse_public_link_url, +}; +use crate::crypto::{decrypt_extended_attributes_unverified, decrypt_link}; use crate::dtos::{ FileDto, FolderChildrenResponse, LinkDetailsDto, LinkDetailsRequest, LinkDetailsResponse, LinkType, PublicLinkAuthRequest, PublicLinkAuthResponse, PublicLinkInfoResponse, - RevisionResponse, + ThumbnailBlockListRequest, ThumbnailBlockListResponse, }; -use crate::node::{Node, NodeKind, RevisionState}; +use crate::node::{FileThumbnail, Node, NodeKind, RevisionState, ThumbnailType}; +use crate::revision::{MAX_CONCURRENT_BLOCK_DOWNLOADS, RevisionReader, decrypt_block_blocking}; use crate::sharing::MemberRole; +use crate::single_flight::SingleFlight; +use crate::transport::{BlockSession, RevisionTransport, rank_block_sizes}; /// Link metadata readable before authenticating, from `GET urls/{token}/info`. /// @@ -69,11 +81,13 @@ pub struct PublicLinkInfo { pub vendor_type: i32, } -/// An authenticated, anonymous session against one public link. +/// Everything needed to run the handshake, and nothing that a completed +/// handshake produced. /// -/// Mirrors TS `SharingPublicLinkSession`. Holds the link password so an expired -/// session can be re-established by replaying the handshake. -struct PublicLinkSession { +/// Split out from [`PublicLinkSession`] so a session can be *constructed* around +/// an already-authenticated client while still being able to replay the +/// handshake later — the credentials outlive any particular session. +struct Credentials { config: ProtonClientConfiguration, token: String, /// URL fragment password concatenated with any custom password — the TS SDK @@ -89,7 +103,7 @@ struct AuthOutcome { public_role: MemberRole, } -impl PublicLinkSession { +impl Credentials { async fn info(&self) -> Result { get_unauthenticated(&self.config, &format!("drive/urls/{}/info", self.token)).await } @@ -159,7 +173,12 @@ impl PublicLinkSession { // scope. Sending them to `drive/` returns 403 `MissingScopes`. // The session routes (`drive/urls/{token}/…`) are exempt — they run // unauthenticated, before this client exists. - .with_base_route("drive/unauth/"); + .with_base_route("drive/unauth/") + // The refresh token above is empty, so `ApiHttpClient`'s built-in 401 + // handling would POST it to `auth/v4/refresh` and replace the real 401 + // with an error about refreshing. Opt out, and let the 401 reach + // `RevisionTransport`, which recovers by replaying *this* handshake. + .without_token_refresh(); Ok(AuthOutcome { http, @@ -190,19 +209,133 @@ impl PublicLinkSession { } } +/// The client a public-link session is currently issuing requests through, +/// tagged with the handshake that produced it. +struct LiveSession { + http: ApiHttpClient, + generation: u64, +} + +/// An authenticated, anonymous session against one public link. +/// +/// Mirrors TS `SharingPublicLinkSession`. Holds the link password so an expired +/// session can be re-established by replaying the handshake — which matters more +/// here than anywhere else in the SDK, because the anonymous session is minted +/// with an **empty refresh token** and so cannot renew itself the ordinary way. +/// A long read (a film, say) will outlive it. +/// +/// The live client is swapped behind a lock rather than handed back to the +/// caller as a new object, so every [`RevisionReader`](crate::RevisionReader) +/// already opened over this session keeps working across the swap. Swapping the +/// *whole* client is required, not just its tokens: a re-auth mints a new +/// `x-pm-uid` as well as a new bearer, and an [`ApiHttpClient`] binds one +/// session id for its lifetime. +struct PublicLinkSession { + credentials: Credentials, + /// Read on every request, written only by a handshake. A `std` lock because + /// [`BlockSession::http`] is synchronous; it is never held across an await, + /// and never poisoned in practice, so poisoning is recovered from rather + /// than propagated. + live: std::sync::RwLock, + /// Serializes handshakes, so a burst of expired-session failures replays one + /// SRP exchange rather than one per caller. + renewing: tokio::sync::Mutex<()>, +} + +impl PublicLinkSession { + /// Authenticate and wrap the result in a renewable session. + /// + /// Returns the outcome alongside the session because the share key, root uid + /// and role belong to the *link*, not to the session, and the client keeps + /// them directly. + async fn open(credentials: Credentials) -> Result<(Arc, AuthOutcome)> { + let auth = credentials.auth().await?; + let session = Arc::new(Self { + credentials, + live: std::sync::RwLock::new(LiveSession { + http: auth.http.clone(), + generation: 0, + }), + renewing: tokio::sync::Mutex::new(()), + }); + Ok((session, auth)) + } + + fn read_live(&self, f: impl FnOnce(&LiveSession) -> T) -> T { + let live = self + .live + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + f(&live) + } +} + +#[async_trait] +impl BlockSession for PublicLinkSession { + fn http(&self) -> (ApiHttpClient, u64) { + self.read_live(|live| (live.http.clone(), live.generation)) + } + + async fn renew(&self, seen: u64) -> Result<()> { + let _guard = self.renewing.lock().await; + + if self.read_live(|live| live.generation) != seen { + // Someone else re-authenticated while we waited; their session is at + // least as fresh as one we would mint now. + return Ok(()); + } + + let auth = self.credentials.auth().await?; + + let mut live = self + .live + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + live.http = auth.http; + live.generation += 1; + Ok(()) + } +} + /// Read access to one public share link, as a visitor. /// /// Obtained from [`open`](Self::open). The client holds an anonymous session and /// the link's share key; every operation is scoped to the shared subtree. +/// +/// Cloning is cheap and shares the session, so a clone recovers from an expiry +/// the original recovered from, and vice versa. +#[derive(Clone)] pub struct ProtonDrivePublicLinkClient { - session: PublicLinkSession, - http: ApiHttpClient, + session: Arc, + /// Every request goes through here rather than through a bare + /// [`ApiHttpClient`], so an expired anonymous session is replayed once and + /// retried instead of surfacing as a 401. + transport: RevisionTransport, /// The link's share key — the parent key of the shared root node. share_key: PrivateKey, root_uid: NodeUid, public_role: MemberRole, + /// Unlocked node keys, so a folder's children do not each re-derive their + /// shared parent's key. + /// + /// Unlocking a key is an S2K derivation — tens of milliseconds. Listing a + /// folder of 500 episodes walked the ancestor chain and re-unlocked every + /// key on it *per child*, which is the difference between a catalog crawl + /// taking seconds and taking minutes. + node_keys: Arc>>, + /// Collapses concurrent misses for the same key onto one load, so the + /// fanned-out detail fetches below do not each start the same ancestor walk + /// before any of them has populated the cache. + key_loads: Arc>, } +/// Unlocked node keys held per client. +/// +/// A shared subtree's *distinct* keys are its folders, so this only has to +/// outlive a crawl of one directory's worth of ancestors; it is a bound on +/// pathological trees, not a working-set estimate. +const NODE_KEY_CACHE_CAPACITY: usize = 512; + impl ProtonDrivePublicLinkClient { /// Read a link's public metadata without opening it. /// @@ -211,12 +344,12 @@ impl ProtonDrivePublicLinkClient { /// its `#password` fragment. pub async fn info(config: ProtonClientConfiguration, url: &str) -> Result { let (token, password) = parse_public_link_url(url)?; - let session = PublicLinkSession { + let credentials = Credentials { config, token, password, }; - let raw = session.info().await?; + let raw = credentials.info().await?; Ok(PublicLinkInfo { is_custom_password_protected: raw.is_custom_password_protected(), is_legacy: raw.is_legacy(), @@ -242,22 +375,50 @@ impl ProtonDrivePublicLinkClient { _ => url_password, }; - let session = PublicLinkSession { + let (session, auth) = PublicLinkSession::open(Credentials { config, token, password, - }; - let auth = session.auth().await?; + }) + .await?; + + let transport = RevisionTransport::new( + session.clone(), + Arc::new(Semaphore::new(DEFAULT_MAX_INFLIGHT_BLOCKS)), + ); Ok(Self { session, - http: auth.http, + transport, share_key: auth.share_key, root_uid: auth.root_uid, public_role: auth.public_role, + node_keys: Arc::new(std::sync::Mutex::new(LruCache::new( + NonZeroUsize::new(NODE_KEY_CACHE_CAPACITY).expect("capacity is non-zero"), + ))), + key_loads: Arc::new(SingleFlight::default()), }) } + /// Cap the content blocks this client keeps in memory at once, across every + /// read it is running. + /// + /// Mirrors + /// [`ProtonDriveClient::with_max_inflight_blocks`](crate::ProtonDriveClient::with_max_inflight_blocks). + /// The default is sized for a background sync; a player that seeks wants it + /// higher. Must be non-zero; a zero cap would deadlock every read, so it is + /// clamped to 1. + /// + /// Applies to this client and every clone made from it *afterwards*; clones + /// already taken keep the previous cap. + pub fn with_max_inflight_blocks(mut self, blocks: usize) -> Self { + self.transport = RevisionTransport::new( + self.session.clone(), + Arc::new(Semaphore::new(blocks.max(1))), + ); + self + } + /// The role the link grants its visitors — [`MemberRole::Viewer`] for a /// read-only link, [`MemberRole::Editor`] for one that permits uploads. pub fn public_role(&self) -> MemberRole { @@ -310,7 +471,7 @@ impl ProtonDrivePublicLinkClient { path.push_str(&format!("?AnchorID={anchor_id}")); } - let page: FolderChildrenResponse = self.http.get(&path).await?; + let page: FolderChildrenResponse = self.transport.get(&path).await?; if page.link_ids.is_empty() { break; } @@ -334,6 +495,10 @@ impl ProtonDrivePublicLinkClient { /// A node that does not exist is omitted; one that fails to decrypt is /// logged and skipped, so the result may be shorter than `uids` — the same /// partial-listing behavior as the authenticated client. + /// + /// Link ids are batched at [`MAX_BATCH_COUNT`] per request — the server's + /// limit, which an unchunked call silently exceeded on any folder past 150 + /// children — and the batches are fetched several at a time. pub async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result> { if uids.is_empty() { return Ok(Vec::new()); @@ -341,28 +506,39 @@ impl ProtonDrivePublicLinkClient { let mut nodes = Vec::with_capacity(uids.len()); for (volume_id, link_ids) in group_by_volume(uids) { - let response: LinkDetailsResponse = self - .http - .post( - &format!("v2/volumes/{volume_id}/links"), - &LinkDetailsRequest { - link_ids: &link_ids, - }, - ) - .await?; + // The closure takes its batch *by value*. Taking `&[LinkId]` — which + // `chunks()` hands out directly — would give the fetch future a + // higher-ranked lifetime that `tokio::spawn` rejects in callers + // ("implementation of `FnOnce` is not general enough"), an error that + // surfaces only downstream. `tests/spawnable.rs` pins it. + let chunks: Vec> = link_ids + .chunks(MAX_BATCH_COUNT) + .map(<[LinkId]>::to_vec) + .collect(); + let batches = chunks.into_iter().map(|link_ids| { + let client = self.clone(); + let volume_id = volume_id.clone(); + async move { client.fetch_link_details(&volume_id, &link_ids).await } + }); - for details in &response.links { - let parent_key = match self.resolve_parent_key(&volume_id, details).await { - Ok(key) => key, - Err(e) => { - tracing::warn!(link_id = %details.link.id, "skipping node: {e}"); - continue; - } - }; - match self.build_node(&volume_id, details, &parent_key) { - Ok(node) => nodes.push(node), - Err(e) => { - tracing::warn!(link_id = %details.link.id, "skipping node: {e}") + // `buffered` yields in request order, so a listing keeps the order + // its uids were given in. + let mut responses = stream::iter(batches).buffered(MAX_CONCURRENT_DETAIL_FETCHES); + + while let Some(response) = responses.try_next().await? { + for details in &response.links { + let parent_key = match self.resolve_parent_key(&volume_id, details).await { + Ok(key) => key, + Err(e) => { + tracing::warn!(link_id = %details.link.id, "skipping node: {e}"); + continue; + } + }; + match self.build_node(&volume_id, details, &parent_key) { + Ok(node) => nodes.push(node), + Err(e) => { + tracing::warn!(link_id = %details.link.id, "skipping node: {e}") + } } } } @@ -371,6 +547,21 @@ impl ProtonDrivePublicLinkClient { Ok(nodes) } + /// One batch of link details. Callers must keep `link_ids` within + /// [`MAX_BATCH_COUNT`]. + async fn fetch_link_details( + &self, + volume_id: &VolumeId, + link_ids: &[LinkId], + ) -> Result { + self.transport + .post( + &format!("v2/volumes/{volume_id}/links"), + &LinkDetailsRequest { link_ids }, + ) + .await + } + /// Download a shared file's active revision. /// /// Unlike the authenticated client's @@ -404,24 +595,35 @@ impl ProtonDrivePublicLinkClient { let decrypted = decrypt_link(&parent_key, &details.link)?; let content_key = self.content_key(&details, &decrypted.node_key)?; - let response: RevisionResponse = self - .http - .get(&format!( - "v2/volumes/{}/files/{}/revisions/{}", - uid.volume_id, uid.link_id, revision_id - )) + // Paged: a single un-paged GET returns only the first 50 blocks, which + // silently truncated any file past 200 MiB. + let (_, blocks) = self + .transport + .list_blocks(&uid.volume_id, &uid.link_id, &revision_id) .await?; - // Blocks arrive in index order; decrypt each in turn straight into the - // writer so a large file never has to be resident in full. - let mut blocks = response.revision.blocks; - blocks.sort_by_key(|block| block.index); - for block in blocks { - let ciphertext = self - .http - .get_storage_blob(&block.bare_url, &block.token) - .await?; - let plaintext = content_key.decrypt_block(&ciphertext)?; + // Fetched and decrypted several at a time but yielded in block order, so + // the writer still sees the file sequentially and no more than + // `MAX_CONCURRENT_BLOCK_DOWNLOADS` blocks are ever resident. + let mut plaintexts = stream::iter(blocks.into_iter().map(|block| { + let transport = self.transport.clone(); + let content_key = content_key.clone(); + async move { + let permit = transport.block_slots().acquire_owned().await.map_err(|e| { + ProtonError::invalid_operation(format!("block slots closed: {e}")) + })?; + let ciphertext = transport + .http() + .get_storage_blob(&block.bare_url, &block.token) + .await?; + let plaintext = decrypt_block_blocking(content_key, ciphertext).await?; + Ok::<_, ProtonError>((plaintext, permit)) + } + })) + .buffered(MAX_CONCURRENT_BLOCK_DOWNLOADS); + + // `_permit` frees once the block is written, not when it is decrypted. + while let Some((plaintext, _permit)) = plaintexts.try_next().await? { writer .write_all(&plaintext) .map_err(|e| ProtonError::invalid_operation(format!("write block: {e}")))?; @@ -430,32 +632,347 @@ impl ProtonDrivePublicLinkClient { Ok(()) } + /// Download and decrypt only the plaintext byte range + /// `[offset, offset + length)` of a shared file's active revision. + /// + /// A one-shot convenience over [`open_revision`](Self::open_revision), which + /// callers issuing more than one read should hold instead — this repeats the + /// whole resolution (link details, ancestor walk, node-key unlock, revision + /// listing) every time. + pub async fn download_range(&self, uid: &NodeUid, offset: u64, length: u64) -> Result> { + self.open_revision(uid).await?.read_at(offset, length).await + } + + /// Open the active revision of a shared file for seekable reads. + /// + /// This is what makes a public link streamable: every content block is a + /// self-contained PGP packet under one session key, so a + /// [`RevisionReader`] can serve any byte range by fetching only the blocks + /// that overlap it, several at a time — no whole-file download, no + /// sequential decrypt. + /// + /// The reader holds this client's session and renews it on expiry, so it + /// keeps working past the anonymous session's lifetime. + /// + /// The reader is pinned to the revision that is active now; it does not + /// follow later revisions of the same file. + pub async fn open_revision(&self, uid: &NodeUid) -> Result { + self.open_revision_inner(uid, None).await + } + + /// As [`open_revision`](Self::open_revision), on an explicit revision. + /// + /// A visitor cannot enumerate a file's revision history, so `revision_id` + /// has to come from somewhere that already knew it — + /// [`NodeKind::File::active_revision_id`](crate::NodeKind) recorded earlier, + /// typically, to reopen the same bytes a previous session was reading. + pub async fn open_revision_at( + &self, + uid: &NodeUid, + revision_id: &str, + ) -> Result { + self.open_revision_inner(uid, Some(revision_id)).await + } + + async fn open_revision_inner( + &self, + uid: &NodeUid, + revision_id: Option<&str>, + ) -> Result { + let details = self.link_details(uid).await?; + let active = file_properties(&details).and_then(|file| file.active_revision.as_ref()); + + let revision_id = match revision_id { + Some(id) => id.to_string(), + None => active.map(|revision| revision.id.clone()).ok_or_else(|| { + ProtonError::invalid_operation("the shared file has no active revision") + })?, + }; + + let parent_key = self.resolve_parent_key(&uid.volume_id, &details).await?; + let node_key = decrypt_link(&parent_key, &details.link)?.node_key; + let content_key = self.content_key(&details, &node_key)?; + + let (revision, blocks) = self + .transport + .list_blocks(&uid.volume_id, &uid.link_id, &revision_id) + .await?; + + // Prefer the revision's own extended attributes. Fall back to the copy + // the link-details response above already delivered for the *active* + // revision — the visitor route has a history of stripping fields the + // authenticated one returns, and without a usable size the reader cannot + // place block boundaries and refuses to open a multi-block file at all. + let armored = revision.extended_attributes.as_deref().or_else(|| { + active + .filter(|candidate| candidate.id == revision_id) + .and_then(|candidate| candidate.extended_attributes.as_deref()) + }); + + let common = armored.and_then(|xattr| { + match decrypt_extended_attributes_unverified(&node_key, xattr) { + Ok(attrs) => attrs.common, + Err(e) => { + tracing::debug!(error = %e, "revision extended attributes did not decrypt"); + None + } + } + }); + + let block_sizes = rank_block_sizes(common.as_ref(), &revision_id, blocks.len())?; + + Ok(RevisionReader::new( + self.transport.clone(), + uid.clone(), + revision_id, + content_key, + blocks, + block_sizes, + )) + } + + /// Download and decrypt a shared file's thumbnail of the given type, if it + /// has one. + /// + /// Ports the authenticated + /// [`download_thumbnail`](crate::ProtonDriveClient::download_thumbnail): the + /// revision's thumbnail header names a block id, `POST volumes/{vid}/thumbnails` + /// resolves it to a download URL, and the bytes decrypt under the same + /// content key as the file's content blocks. Only the route prefix differs. + /// Returns `Ok(None)` when the file has no thumbnail of that type. + pub async fn download_thumbnail( + &self, + uid: &NodeUid, + thumbnail_type: ThumbnailType, + ) -> Result>> { + let (content_key, thumbnail_id) = match self.thumbnail_target(uid, thumbnail_type).await? { + (content_key, Some(id)) => (content_key, id), + (_, None) => return Ok(None), + }; + + let response: ThumbnailBlockListResponse = self + .transport + .post( + &format!("volumes/{}/thumbnails", uid.volume_id), + &ThumbnailBlockListRequest { + thumbnail_ids: vec![thumbnail_id.clone()], + }, + ) + .await?; + + let block = response + .blocks + .into_iter() + .find(|block| block.thumbnail_id == thumbnail_id) + .ok_or_else(|| { + let reason = response + .errors + .iter() + .find(|e| e.thumbnail_id == thumbnail_id) + .map(|e| e.error.clone()) + .unwrap_or_else(|| "thumbnail block not returned".to_string()); + ProtonError::invalid_operation(format!("resolve thumbnail block: {reason}")) + })?; + + let ciphertext = self + .transport + .http() + .get_storage_blob(&block.bare_url, &block.token) + .await?; + Ok(Some(content_key.decrypt_thumbnail(&ciphertext)?)) + } + + /// Batch-download the thumbnails of `uids`. + /// + /// The shape a poster wall wants: block ids are resolved in batches of up to + /// [`MAX_THUMBNAIL_IDS_PER_REQUEST`], and a per-file failure — no thumbnail + /// of that type, a decrypt error, a node that is not a file — is reported in + /// that file's [`FileThumbnail`] instead of failing the batch. Returned + /// order does not track the input order. + pub async fn enumerate_thumbnails( + &self, + uids: &[NodeUid], + thumbnail_type: ThumbnailType, + ) -> Result> { + let mut results: Vec = Vec::new(); + + for (volume_id, link_ids) in group_by_volume(uids) { + // thumbnail id -> the file it belongs to and the key it decrypts under. + let mut targets: HashMap = HashMap::new(); + + for link_id in link_ids { + let uid = NodeUid::new(volume_id.clone(), link_id); + match self.thumbnail_target(&uid, thumbnail_type).await { + Ok((content_key, Some(id))) => { + targets.insert(id, (uid, content_key)); + } + Ok((_, None)) => results.push(FileThumbnail::err( + uid.clone(), + ProtonError::invalid_operation(format!( + "node {uid} has no thumbnail of the requested type" + )), + )), + Err(e) => results.push(FileThumbnail::err(uid, e)), + } + } + + let thumbnail_ids: Vec = targets.keys().cloned().collect(); + for chunk in thumbnail_ids.chunks(MAX_THUMBNAIL_IDS_PER_REQUEST) { + self.drain_thumbnail_chunk(&volume_id, chunk, &mut targets, &mut results) + .await; + } + } + + Ok(results) + } + + /// Resolve one chunk of thumbnail ids to blocks, download them, and record + /// an outcome for every id in the chunk — including the ones the server said + /// nothing about at all. + async fn drain_thumbnail_chunk( + &self, + volume_id: &VolumeId, + chunk: &[String], + targets: &mut HashMap, + results: &mut Vec, + ) { + let response: ThumbnailBlockListResponse = match self + .transport + .post( + &format!("volumes/{volume_id}/thumbnails"), + &ThumbnailBlockListRequest { + thumbnail_ids: chunk.to_vec(), + }, + ) + .await + { + Ok(response) => response, + // The chunk request itself failed: every file in it gets that error. + Err(e) => { + for id in chunk { + if let Some((uid, _)) = targets.remove(id) { + results.push(FileThumbnail::err( + uid, + ProtonError::invalid_operation(format!( + "resolve thumbnail blocks: {e}" + )), + )); + } + } + return; + } + }; + + for block in response.blocks { + let Some((uid, content_key)) = targets.remove(&block.thumbnail_id) else { + continue; + }; + let downloaded = match self + .transport + .http() + .get_storage_blob(&block.bare_url, &block.token) + .await + { + Ok(ciphertext) => content_key + .decrypt_thumbnail(&ciphertext) + .map_err(ProtonError::from), + Err(e) => Err(e), + }; + results.push(match downloaded { + Ok(bytes) => FileThumbnail::ok(uid, bytes), + Err(e) => FileThumbnail::err(uid, e), + }); + } + + for error in response.errors { + if let Some((uid, _)) = targets.remove(&error.thumbnail_id) { + results.push(FileThumbnail::err( + uid, + ProtonError::invalid_operation(error.error), + )); + } + } + + // Anything still in `targets` was neither returned nor errored on. Left + // unreported it would look like a file that was never asked about. + for id in chunk { + if let Some((uid, _)) = targets.remove(id) { + results.push(FileThumbnail::err( + uid, + ProtonError::invalid_operation("thumbnail not found".to_string()), + )); + } + } + } + + /// A file's content key and the block id of its thumbnail of + /// `thumbnail_type`, if it has one. + async fn thumbnail_target( + &self, + uid: &NodeUid, + thumbnail_type: ThumbnailType, + ) -> Result<(ContentKey, Option)> { + let details = self.link_details(uid).await?; + let revision_id = file_properties(&details) + .and_then(|file| file.active_revision.as_ref()) + .map(|revision| revision.id.clone()) + .ok_or_else(|| { + ProtonError::invalid_operation("the shared file has no active revision") + })?; + + let parent_key = self.resolve_parent_key(&uid.volume_id, &details).await?; + let node_key = decrypt_link(&parent_key, &details.link)?.node_key; + let content_key = self.content_key(&details, &node_key)?; + + // The thumbnail headers ride along with the revision's block listing. + let (revision, _blocks) = self + .transport + .list_blocks(&uid.volume_id, &uid.link_id, &revision_id) + .await?; + let wanted = thumbnail_type.as_i32(); + let thumbnail_id = revision + .thumbnails + .iter() + .find(|thumbnail| thumbnail.thumbnail_type == wanted) + .and_then(|thumbnail| thumbnail.id.clone()); + + Ok((content_key, thumbnail_id)) + } + /// Re-run the handshake, replacing the session behind this client. /// /// The anonymous session carries no refresh token, so an expiry is recovered /// from by authenticating again (TS `SharingPublicLinkSession.reauth`). - /// Returns a fresh client rather than mutating this one, so an in-flight - /// caller never observes a half-swapped session. + /// + /// Rarely needed directly: every request this client makes already replays + /// the handshake once on an expired session, and so does every + /// [`RevisionReader`](crate::RevisionReader) it hands out. Reach for it to + /// renew *ahead* of a long read rather than to repair a failed one. + /// + /// The share key, root uid and role belong to the link and cannot change + /// under a successful re-auth, so only the session is replaced. + pub async fn refresh_session(&self) -> Result<()> { + let (_, generation) = self.session.http(); + self.session.renew(generation).await + } + + /// As [`refresh_session`](Self::refresh_session), returning a client for + /// callers written against the older shape. + /// + /// The returned client shares this one's session rather than owning a + /// separate one — which is the point of the change: readers opened from + /// *either* handle recover from the expiry, where previously the original + /// client and everything it had handed out were stranded. pub async fn reauth(&self) -> Result { - let auth = self.session.auth().await?; - Ok(Self { - session: PublicLinkSession { - config: self.session.config.clone(), - token: self.session.token.clone(), - password: self.session.password.clone(), - }, - http: auth.http, - share_key: auth.share_key, - root_uid: auth.root_uid, - public_role: auth.public_role, - }) + self.refresh_session().await?; + Ok(self.clone()) } /// The key that decrypts `link`: the share key at the shared root, otherwise - /// the parent node's key, resolved recursively up to the root. + /// the parent node's key. /// - /// The recursion terminates at the shared root because the visitor cannot - /// see past it — a link whose parent chain does not reach the root is not + /// The walk terminates at the shared root because the visitor cannot see + /// past it — a link whose parent chain does not reach the root is not /// readable with this session's material. async fn resolve_parent_key( &self, @@ -470,24 +987,88 @@ impl ProtonDrivePublicLinkClient { ProtonError::invalid_operation("node has no parent and is not the shared root") })?; - let parent = self - .link_details(&NodeUid::new(volume_id.clone(), parent_id)) - .await?; - // Recursion is boxed: the chain length is the tree depth, which the - // compiler cannot bound for an `async fn` that calls itself. - let grandparent_key = Box::pin(self.resolve_parent_key(volume_id, &parent)).await?; - Ok(decrypt_link(&grandparent_key, &parent.link)?.node_key) + self.node_key(&NodeUid::new(volume_id.clone(), parent_id)) + .await + } + + /// A node's own unlocked key, from cache when it is there. + /// + /// Walks *up* to the nearest cached ancestor (or the shared root, whose + /// parent key is the share key), then unlocks back *down*, caching every key + /// on the way. Written iteratively rather than recursively so it is one + /// finite future — an `async fn` that awaited itself would need boxing at + /// the recursion point, and its `Send`-ness would depend on its own, which + /// the compiler cannot resolve. + async fn node_key(&self, uid: &NodeUid) -> Result { + if let Some(key) = self.cached_node_key(uid) { + return Ok(key); + } + + let client = self.clone(); + let target = uid.clone(); + self.key_loads + .run(uid.clone(), async move { + // Up: collect the chain from `target` towards the root, stopping + // at the first key already unlocked. + let mut chain: Vec = Vec::new(); + let mut cursor = target.clone(); + + let mut key = loop { + if let Some(key) = client.cached_node_key(&cursor) { + break key; + } + + let details = client.link_details(&cursor).await?; + + if details.link.id == client.root_uid.link_id { + let key = decrypt_link(&client.share_key, &details.link)?.node_key; + client.cache_node_key(&cursor, &key); + break key; + } + + let parent_id = details.link.parent_id.clone().ok_or_else(|| { + ProtonError::invalid_operation( + "node has no parent and is not the shared root", + ) + })?; + let parent = NodeUid::new(cursor.volume_id.clone(), parent_id); + chain.push(details); + cursor = parent; + }; + + // Down: each node's key unlocks the next, and every one is worth + // keeping — the siblings still to be listed will ask for them. + while let Some(details) = chain.pop() { + key = decrypt_link(&key, &details.link)?.node_key; + client.cache_node_key( + &NodeUid::new(target.volume_id.clone(), details.link.id.clone()), + &key, + ); + } + + Ok(key) + }) + .await + } + + fn cached_node_key(&self, uid: &NodeUid) -> Option { + self.node_keys + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(uid) + .cloned() + } + + fn cache_node_key(&self, uid: &NodeUid, key: &PrivateKey) { + self.node_keys + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .put(uid.clone(), key.clone()); } async fn link_details(&self, uid: &NodeUid) -> Result { - let response: LinkDetailsResponse = self - .http - .post( - &format!("v2/volumes/{}/links", uid.volume_id), - &LinkDetailsRequest { - link_ids: std::slice::from_ref(&uid.link_id), - }, - ) + let response = self + .fetch_link_details(&uid.volume_id, std::slice::from_ref(&uid.link_id)) .await?; response .links @@ -522,20 +1103,42 @@ impl ProtonDrivePublicLinkClient { NodeKind::Folder } else { let file = file_properties(details); + let active = file.and_then(|f| f.active_revision.as_ref()); + + // The link-details response already carries the active revision's + // `XAttr`, so the uploader's claimed size, mtime and digest cost + // nothing beyond one decrypt — no extra round-trip. Unverified, + // because a visitor has no address book to resolve the signer + // through; see the module note. + let common = active + .and_then(|revision| revision.extended_attributes.as_deref()) + .and_then(|xattr| { + match decrypt_extended_attributes_unverified(&decrypted.node_key, xattr) { + Ok(attrs) => attrs.common, + // Present but unreadable is worth a line — absent is + // ordinary, and says nothing. + Err(e) => { + tracing::debug!(link_id = %link.id, error = %e, + "revision extended attributes did not decrypt"); + None + } + } + }); + NodeKind::File { media_type: file.map(|f| f.media_type.clone()).unwrap_or_default(), total_size_on_storage: file.map(|f| f.total_size_on_storage).unwrap_or_default(), - active_revision_state: file - .and_then(|f| f.active_revision.as_ref()) + active_revision_state: active .map(|revision| RevisionState::from_raw(revision.state)), - active_revision_id: file - .and_then(|f| f.active_revision.as_ref()) - .map(|revision| revision.id.clone()), - // Claimed metadata lives in the revision's extended attributes, - // which this read surface does not fetch. - claimed_size: None, - claimed_modification_time: None, - content_sha1: None, + active_revision_id: active.map(|revision| revision.id.clone()), + claimed_size: common.as_ref().and_then(|c| c.size), + claimed_modification_time: common + .as_ref() + .and_then(|c| c.modification_time.clone()), + content_sha1: common + .as_ref() + .and_then(|c| c.digests.as_ref()) + .and_then(|d| d.sha1.clone()), } }; diff --git a/crates/proton-drive-rs/src/revision.rs b/crates/proton-drive-rs/src/revision.rs index e0eaa0a..f799fc5 100644 --- a/crates/proton-drive-rs/src/revision.rs +++ b/crates/proton-drive-rs/src/revision.rs @@ -14,8 +14,8 @@ use crate::node::RevisionState; use sha2::{Digest, Sha256}; use tokio::sync::{Mutex, OwnedSemaphorePermit, RwLock}; -use crate::client::ProtonDriveClient; use crate::dtos::BlockDto; +use crate::transport::RevisionTransport; /// Content blocks fetched concurrently within a single file. /// @@ -134,7 +134,11 @@ struct BlockTable { /// file gains a new revision, this reader keeps serving the old one; reopen to /// follow the change. pub struct RevisionReader { - client: ProtonDriveClient, + /// Everything the read needs from whoever opened it: block permits, an HTTP + /// client already carrying the right route prefix, and — for a session that + /// can die under the reader — the renewal path. Deliberately *not* a whole + /// client, so an anonymous public-link visitor can have one too. + transport: RevisionTransport, uid: NodeUid, revision_id: String, content_key: ContentKey, @@ -149,7 +153,7 @@ pub struct RevisionReader { impl RevisionReader { pub(crate) fn new( - client: ProtonDriveClient, + transport: RevisionTransport, uid: NodeUid, revision_id: String, content_key: ContentKey, @@ -158,7 +162,7 @@ impl RevisionReader { ) -> Self { let file_size = block_sizes.iter().sum(); Self { - client, + transport, uid, revision_id, content_key, @@ -249,7 +253,7 @@ impl RevisionReader { // Held across fetch *and* decrypt: both halves are resident 4 MiB // buffers, and the plaintext outlives the ciphertext. let permit = self - .client + .transport .block_slots() .acquire_owned() .await @@ -257,12 +261,14 @@ impl RevisionReader { let (url, token, generation) = self.block_location(index).await?; - let ciphertext = match self.client.http().get_storage_blob(&url, &token).await { + // A block fetch carries a `pm-storage-token` and no session credential + // at all, so it cannot fail for session reasons — only the URL expires. + let ciphertext = match self.transport.http().get_storage_blob(&url, &token).await { Ok(bytes) => bytes, Err(e) if is_expired_block_url(&e) => { self.refresh_blocks(generation).await?; let (url, token, _) = self.block_location(index).await?; - self.client.http().get_storage_blob(&url, &token).await? + self.transport.http().get_storage_blob(&url, &token).await? } Err(e) => return Err(e), }; @@ -300,8 +306,8 @@ impl RevisionReader { } let (_, blocks) = self - .client - .fetch_revision_blocks(&self.uid.volume_id, &self.uid.link_id, &self.revision_id) + .transport + .list_blocks(&self.uid.volume_id, &self.uid.link_id, &self.revision_id) .await?; if blocks.len() != self.block_sizes.len() { diff --git a/crates/proton-drive-rs/src/transport.rs b/crates/proton-drive-rs/src/transport.rs new file mode 100644 index 0000000..dd293d8 --- /dev/null +++ b/crates/proton-drive-rs/src/transport.rs @@ -0,0 +1,492 @@ +//! What a revision read needs from its client, and nothing else. +//! +//! [`RevisionReader`](crate::RevisionReader) originally held a whole +//! [`ProtonDriveClient`](crate::ProtonDriveClient), which is why an anonymous +//! public-link visitor could not have one. It only ever used three things: the +//! host-wide block permits, an [`ApiHttpClient`] to fetch block bodies through, +//! and the paginated revision listing. All three are satisfied by an HTTP client +//! plus a semaphore — and `ApiHttpClient` already carries the route prefix +//! (`drive/` vs `drive/unauth/`), which is the only thing that differs between +//! the authenticated and visitor paths. The endpoints, payloads, paging and +//! contiguity rules are identical. +//! +//! So the seam here is deliberately narrow: [`RevisionTransport`] is *concrete*, +//! and the one genuine axis of variation — what to do when the session dies — +//! is the two-method [`BlockSession`] trait. +//! +//! ## The session asymmetry +//! +//! Worth stating once, because it is easy to re-derive wrongly: block bodies are +//! fetched with [`ApiHttpClient::get_storage_blob`], which sends a +//! `pm-storage-token` and **no bearer and no `x-pm-uid`**. Block fetches are +//! therefore immune to session expiry — a 401 there means an expired *block +//! URL*. The revision listing is the only session-authenticated call in a read, +//! so it is the only one that can meet a dead session. + +use std::sync::Arc; + +use async_trait::async_trait; +use proton_sdk::error::{ProtonError, Result}; +use proton_sdk::http::ApiHttpClient; +use proton_sdk::ids::{LinkId, VolumeId}; +use serde::Serialize; +use serde::de::DeserializeOwned; +use tokio::sync::Semaphore; + +use crate::dtos::{BlockDto, DecryptedCommonExtendedAttributes, RevisionDto, RevisionResponse}; + +/// Content blocks are 4 MiB of plaintext each (C# `RevisionWriter.DefaultBlockSize`). +pub(crate) const DEFAULT_BLOCK_SIZE: usize = 1 << 22; + +/// The live session a revision read issues its authenticated calls through, and +/// how to replace it when it dies. +/// +/// Generation-versioned for the same reason +/// [`RevisionReader::refresh_blocks`](crate::RevisionReader) versions its block +/// table: a burst of failures must replay **one** recovery, not one per caller. +/// A loser re-checks the generation and returns. +#[async_trait] +pub(crate) trait BlockSession: Send + Sync + 'static { + /// The client to use right now, paired with the generation it belongs to. + /// + /// Deliberately synchronous — this is on the per-request path, and it is a + /// lock read plus two refcount bumps. The lock is never held across an await. + fn http(&self) -> (ApiHttpClient, u64); + + /// Replace the session, unless somebody has already replaced `seen`. + async fn renew(&self, seen: u64) -> Result<()>; +} + +/// A session that renews itself. +/// +/// An ordinary bearer session refreshes its own tokens inside [`ApiHttpClient`] +/// on a 401, so by the time an error surfaces here there is nothing left to try: +/// `renew` is a no-op and the generation never moves. +pub(crate) struct StaticSession(ApiHttpClient); + +impl StaticSession { + pub(crate) fn new(http: ApiHttpClient) -> Self { + Self(http) + } +} + +#[async_trait] +impl BlockSession for StaticSession { + fn http(&self) -> (ApiHttpClient, u64) { + (self.0.clone(), 0) + } + + async fn renew(&self, _seen: u64) -> Result<()> { + Ok(()) + } +} + +/// Everything a revision read needs that is not the revision itself. +#[derive(Clone)] +pub(crate) struct RevisionTransport { + session: Arc, + /// Host-wide in-flight block permits, shared with the owning client's other + /// transfers. + block_slots: Arc, +} + +impl RevisionTransport { + pub(crate) fn new(session: Arc, block_slots: Arc) -> Self { + Self { + session, + block_slots, + } + } + + /// A transport over a session that manages its own tokens. + pub(crate) fn authenticated(http: ApiHttpClient, block_slots: Arc) -> Self { + Self::new(Arc::new(StaticSession::new(http)), block_slots) + } + + /// The current HTTP client. Block fetches go straight through this — they + /// carry no session credential, so they need no renewal path. + pub(crate) fn http(&self) -> ApiHttpClient { + self.session.http().0 + } + + pub(crate) fn block_slots(&self) -> Arc { + self.block_slots.clone() + } + + /// `GET path` through the current session, replaying the handshake once if + /// the session turns out to be gone. + pub(crate) async fn get(&self, path: &str) -> Result { + let (http, generation) = self.session.http(); + match http.get::(path).await { + Err(e) if is_session_expired(&e) => { + self.session.renew(generation).await?; + self.session.http().0.get::(path).await + } + other => other, + } + } + + /// `POST path` through the current session, with the same one-shot renewal. + pub(crate) async fn post( + &self, + path: &str, + body: &B, + ) -> Result { + let (http, generation) = self.session.http(); + match http.post::(path, body).await { + Err(e) if is_session_expired(&e) => { + self.session.renew(generation).await?; + self.session.http().0.post::(path, body).await + } + other => other, + } + } + + /// The revision's metadata and its full, ordered block table. + pub(crate) async fn list_blocks( + &self, + volume_id: &VolumeId, + link_id: &LinkId, + revision_id: &str, + ) -> Result<(RevisionDto, Vec)> { + const PAGE_SIZE: i32 = 50; + + let mut blocks: Vec = Vec::new(); + let mut metadata: Option = None; + let mut from_index: i32 = 1; + + loop { + let path = format!( + "v2/volumes/{volume_id}/files/{link_id}/revisions/{revision_id}?FromBlockIndex={from_index}&PageSize={PAGE_SIZE}&NoBlockUrls=0" + ); + let response: RevisionResponse = self.get(&path).await?; + let mut revision = response.revision; + let page = std::mem::take(&mut revision.blocks); + let page_len = page.len(); + + if metadata.is_none() { + metadata = Some(revision); + } + blocks.extend(page); + + if page_len < PAGE_SIZE as usize { + break; + } + from_index = blocks.iter().map(|b| b.index).max().unwrap_or(from_index) + 1; + } + + blocks.sort_by_key(|b| b.index); + for (offset, block) in blocks.iter().enumerate() { + if block.index != offset as i32 + 1 { + return Err(ProtonError::invalid_operation( + "file contents are incomplete (non-contiguous blocks)", + )); + } + } + + let metadata = metadata + .ok_or_else(|| ProtonError::invalid_operation("revision returned no metadata"))?; + Ok((metadata, blocks)) + } +} + +/// Whether a failure means the *session* is gone. +/// +/// Only meaningful for session-authenticated calls. A block fetch carries a +/// `pm-storage-token` rather than the bearer, so a 401 there is an expired block +/// URL — see `revision::is_expired_block_url`, which deliberately also accepts +/// 403 and 404. +/// +/// 403 is excluded here on purpose: on the visitor path it means +/// `MissingScopes`, i.e. the request went to `drive/` instead of +/// `drive/unauth/`. Re-authenticating would not fix that and would loop. +pub(crate) fn is_session_expired(error: &ProtonError) -> bool { + matches!(error, ProtonError::Api(e) if e.http_status == 401) +} + +/// Rank a revision's block-size sources and produce the per-block plaintext +/// sizes. +/// +/// These sizes are not merely descriptive: [`RevisionReader`](crate::RevisionReader) +/// accumulates them to derive where each block *starts* in the plaintext. A +/// vector that overstates any block but the last therefore shifts every block +/// after it, and the reader splices those blocks at the wrong offsets — +/// returning full-length reads of the wrong bytes, with no error to notice. See +/// `revision.rs::padded_block_sizes_shift_later_blocks`, which pins that +/// consequence. +/// +/// So the sources are ranked by how much they can be trusted, and this refuses +/// rather than guesses when none of them holds: +/// +/// 1. `Common.BlockSizes` from the revision's extended attributes — the +/// authoritative value, written by the uploading client. +/// 2. `Common.Size` — the total, from which all-but-last full blocks and a final +/// short block follow by subtraction. Sound for the standard layout the +/// uploader produces. +/// 3. Nothing usable. A single-block file cannot be misaligned (there is no +/// later block to shift), so it is assumed full-size and only its reported +/// length can be wrong. A multi-block file is an error: serving it would mean +/// serving misplaced bytes. +/// +/// **Not a rung:** `FileDto.total_size_on_storage` (wire `TotalEncryptedSize`). +/// That is the *ciphertext* size — PGP framing per block, varying with the +/// encryption mode — so deriving plaintext block boundaries from it produces +/// exactly the shifted-splice failure above. Do not add it. +pub(crate) fn rank_block_sizes( + common: Option<&DecryptedCommonExtendedAttributes>, + revision_id: &str, + block_count: usize, +) -> Result> { + let block = DEFAULT_BLOCK_SIZE as u64; + + if let Some(sizes) = common.and_then(|c| c.block_sizes.as_ref()) + && sizes.len() == block_count + { + // A non-positive entry is as corrupting as a padded one — a zero-length + // block in the middle shifts everything after it — so a malformed vector + // is rejected outright rather than clamped. + if let Some(bad) = sizes.iter().position(|&n| n <= 0) { + return Err(ProtonError::invalid_operation(format!( + "revision {revision_id} records a non-positive size for block {bad}" + ))); + } + return Ok(sizes.iter().map(|&n| n as u64).collect()); + } + + if let Some(total) = common.and_then(|c| c.size).filter(|&n| n >= 0) { + let total = total as u64; + return Ok((0..block_count) + .map(|i| total.saturating_sub(block * i as u64).min(block)) + .collect()); + } + + if block_count <= 1 { + return Ok(vec![block; block_count]); + } + + Err(ProtonError::invalid_operation(format!( + "revision {revision_id} has {block_count} blocks but no usable size information \ + (no block sizes and no total in its extended attributes)" + ))) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + fn common( + size: Option, + block_sizes: Option>, + ) -> DecryptedCommonExtendedAttributes { + DecryptedCommonExtendedAttributes { + size, + block_sizes, + ..Default::default() + } + } + + const MIB: u64 = 1 << 20; + + /// The authoritative rung: whatever the uploader recorded is used as-is, + /// including a short final block. + #[test] + fn block_sizes_from_extended_attributes_are_used_verbatim() { + let attrs = common(Some(9 * MIB as i64), Some(vec![4 << 20, 4 << 20, 1 << 20])); + let sizes = rank_block_sizes(Some(&attrs), "rev", 3).expect("sizes resolve"); + assert_eq!(sizes, vec![4 * MIB, 4 * MIB, MIB]); + } + + /// A vector that does not describe every block is not partial information, + /// it is untrustworthy information — fall through to the total rather than + /// pad it out. + #[test] + fn a_block_sizes_vector_of_the_wrong_length_falls_through_to_the_total() { + let attrs = common(Some(9 * MIB as i64), Some(vec![4 << 20, 4 << 20])); + let sizes = rank_block_sizes(Some(&attrs), "rev", 3).expect("sizes resolve"); + assert_eq!(sizes, vec![4 * MIB, 4 * MIB, MIB], "derived from the total"); + } + + /// A zero-length block in the middle shifts every block after it just as a + /// padded one does, so it is refused rather than clamped to something + /// plausible. + #[test] + fn a_non_positive_block_size_is_rejected_rather_than_clamped() { + let attrs = common(Some(9 * MIB as i64), Some(vec![4 << 20, 0, 1 << 20])); + let error = rank_block_sizes(Some(&attrs), "rev", 3).expect_err("must refuse"); + assert!( + error.to_string().contains("non-positive size for block 1"), + "names the offending block: {error}" + ); + } + + /// The total alone is enough, because the uploader's layout is all-but-last + /// full blocks plus a remainder. + #[test] + fn the_total_size_derives_full_blocks_and_a_short_final_one() { + let attrs = common(Some(9 * MIB as i64), None); + let sizes = rank_block_sizes(Some(&attrs), "rev", 3).expect("sizes resolve"); + assert_eq!(sizes, vec![4 * MIB, 4 * MIB, MIB]); + assert_eq!(sizes.iter().sum::(), 9 * MIB, "sums to the total"); + } + + /// A file that happens to be an exact multiple of the block size gets no + /// short tail, and no phantom extra block. + #[test] + fn a_total_that_is_a_whole_number_of_blocks_has_no_short_tail() { + let attrs = common(Some(8 * MIB as i64), None); + let sizes = rank_block_sizes(Some(&attrs), "rev", 2).expect("sizes resolve"); + assert_eq!(sizes, vec![4 * MIB, 4 * MIB]); + } + + /// With one block there is no later block to displace, so assuming the full + /// block size can only misreport the length — never misplace bytes. + #[test] + fn a_single_block_revision_is_assumed_full_size() { + let sizes = rank_block_sizes(None, "rev", 1).expect("sizes resolve"); + assert_eq!(sizes, vec![4 * MIB]); + assert_eq!( + rank_block_sizes(None, "rev", 0).expect("empty resolves"), + Vec::::new() + ); + } + + /// The refusal that keeps the reader honest: guessing here would serve + /// full-length reads of the wrong bytes. + #[test] + fn a_multi_block_revision_with_no_size_information_is_an_error() { + let error = rank_block_sizes(None, "rev-42", 3).expect_err("must refuse"); + let message = error.to_string(); + assert!(message.contains("rev-42"), "names the revision: {message}"); + assert!( + message.contains("no usable size information"), + "says why: {message}" + ); + } + + /// A dead session is a 401 and nothing else. 403 on the visitor path means + /// `MissingScopes` — a wrong route prefix, which re-authenticating would + /// only loop on — and 404 means an expired block URL. + #[test] + fn only_a_401_reads_as_an_expired_session() { + use proton_sdk::api::ResponseCode; + use proton_sdk::error::ProtonApiError; + + let api = |status: u16| { + ProtonError::Api(ProtonApiError { + code: ResponseCode::Unknown, + http_status: status, + message: "nope".into(), + details: None, + }) + }; + + assert!(is_session_expired(&api(401))); + assert!( + !is_session_expired(&api(403)), + "MissingScopes is not expiry" + ); + assert!( + !is_session_expired(&api(404)), + "an expired block url is not" + ); + assert!(!is_session_expired(&ProtonError::invalid_operation("x"))); + } + + /// A session that counts its handshakes, so renewal can be tested without a + /// network. + struct CountingSession { + http: ApiHttpClient, + generation: std::sync::Mutex, + handshakes: AtomicUsize, + } + + #[async_trait] + impl BlockSession for CountingSession { + fn http(&self) -> (ApiHttpClient, u64) { + (self.http.clone(), *self.generation.lock().unwrap()) + } + + async fn renew(&self, seen: u64) -> Result<()> { + let mut generation = self.generation.lock().unwrap(); + if *generation != seen { + // Somebody else already replaced the session we were told about. + return Ok(()); + } + self.handshakes.fetch_add(1, Ordering::SeqCst); + *generation += 1; + Ok(()) + } + } + + fn counting_session() -> Arc { + use proton_sdk::config::ProtonClientConfiguration; + use proton_sdk::http::Tokens; + use proton_sdk::ids::SessionId; + + let http = ApiHttpClient::new( + ProtonClientConfiguration::new("test@1.0"), + SessionId::from("anonymous"), + Tokens { + access_token: "access".into(), + refresh_token: String::new(), + }, + ) + .expect("build client"); + + Arc::new(CountingSession { + http, + generation: std::sync::Mutex::new(0), + handshakes: AtomicUsize::new(0), + }) + } + + /// Concurrent readers that all saw the same dead session replay **one** + /// handshake between them — the generation check makes the losers no-ops. + /// Without this, a seek that spans ten blocks would run ten SRP handshakes. + #[tokio::test] + async fn concurrent_renewals_of_one_generation_run_a_single_handshake() { + let session = counting_session(); + + let mut set = tokio::task::JoinSet::new(); + for _ in 0..8 { + let session = session.clone(); + set.spawn(async move { session.renew(0).await }); + } + while let Some(result) = set.join_next().await { + result.expect("task").expect("renew"); + } + + assert_eq!( + session.handshakes.load(Ordering::SeqCst), + 1, + "eight callers, one handshake" + ); + assert_eq!(*session.generation.lock().unwrap(), 1); + } + + /// A caller arriving with a stale generation has already been rescued by + /// somebody else's handshake, so it must not start another. + #[tokio::test] + async fn renewing_a_generation_that_was_already_replaced_is_a_no_op() { + let session = counting_session(); + + session.renew(0).await.expect("first renew"); + assert_eq!(session.handshakes.load(Ordering::SeqCst), 1); + + session.renew(0).await.expect("stale renew"); + assert_eq!( + session.handshakes.load(Ordering::SeqCst), + 1, + "the stale caller did not re-handshake" + ); + } + + /// The transport is what `RevisionReader` stores, so it has to survive being + /// moved into a spawned task alongside the reader. + #[test] + fn a_revision_transport_is_send_and_sync() { + fn assert_send_sync() {} + assert_send_sync::(); + } +} diff --git a/crates/proton-drive-rs/tests/live_sharing.rs b/crates/proton-drive-rs/tests/live_sharing.rs index b8f4435..817e681 100644 --- a/crates/proton-drive-rs/tests/live_sharing.rs +++ b/crates/proton-drive-rs/tests/live_sharing.rs @@ -19,7 +19,8 @@ mod common; use proton_drive_rs::{ - MemberRole, ProtonDriveClient, ProtonDrivePublicLinkClient, ProtonPhotosClient, + MemberRole, ProtonDriveClient, ProtonDrivePublicLinkClient, ProtonPhotosClient, Thumbnail, + ThumbnailType, }; use proton_sdk::ids::NodeUid; @@ -439,3 +440,318 @@ async fn custom_password_link_requires_the_password() { cleanup(client, &folder).await; } + +// --------------------------------------------------------------------------- +// Public link — seekable reads, the streaming path +// --------------------------------------------------------------------------- + +/// A multi-block file, in blocks of `4 MiB`, whose every byte encodes its own +/// offset. A range served from the wrong block is then identifiable rather than +/// merely unequal — which matters here, because a wrong block-size vector +/// returns a full-length read of the wrong bytes with no error at all. +fn positional_payload(len: usize) -> Vec { + (0..len).map(|i| (i % 251) as u8).collect() +} + +/// **The load-bearing test for streaming over a public link.** +/// +/// Everything `proton-stream` does rests on one uncertain fact: that the +/// `drive/unauth/` revision endpoint returns a usable `XAttr` to an anonymous +/// visitor. Without it `rank_block_sizes` cannot place block boundaries and +/// refuses to open a multi-block file, and seeking is impossible. The visitor +/// route has form here — `?NoBlockUrls=true` already strips `BareURL`/`Token` — +/// so this is verified, not assumed. +/// +/// Then the reads themselves: a range spanning a block boundary, a range wholly +/// inside a later block (the seek case), and the short final block. +#[tokio::test] +#[ignore = "live: needs test-account credentials"] +async fn a_visitor_can_seek_within_a_shared_multi_block_file() { + let Some(live) = common::live_client().await else { + return; + }; + let client = &live.client; + + let folder = scratch_folder(client, "public-seek").await; + + // 9 MiB: two full 4 MiB blocks and a 1 MiB tail, so both the boundary case + // and the short-tail case are reachable. + const BLOCK: usize = 4 << 20; + let contents = positional_payload(9 << 20); + let name = format!("seekable-{}.bin", common::unique_suffix()); + let file = client + .upload_file(&folder, &name, "application/octet-stream", &contents) + .await + .expect("upload_file"); + + let link = client + .create_public_link(&folder, MemberRole::Viewer, None, None) + .await + .expect("create_public_link"); + let url = link.url.clone().expect("a created link must carry its URL"); + + let visitor = ProtonDrivePublicLinkClient::open(common::config(), &url, None) + .await + .expect("open public link"); + + let reader = visitor + .open_revision(&file) + .await + .expect("a visitor must be able to open a revision for reading"); + + // If this fails, the unauth revision response carried no usable size and the + // whole streaming design needs revisiting — so it is asserted before the + // reads, with a message that says which rung was missing. + assert_eq!( + reader.block_sizes(), + &[BLOCK as u64, BLOCK as u64, 1 << 20], + "the visitor must resolve true per-block sizes; a padded or empty vector \ + means the unauth revision response omitted its XAttr" + ); + assert_eq!( + reader.size(), + contents.len() as u64, + "and they must sum to the real file size" + ); + + // A range straddling the block 0 / block 1 boundary. + let straddle = reader + .read_at(BLOCK as u64 - 1024, 2048) + .await + .expect("read across a block boundary"); + assert_eq!( + straddle, + &contents[BLOCK - 1024..BLOCK + 1024], + "a read spanning two blocks must splice them at the right offset" + ); + + // The seek case: a range wholly inside the *last* full block, touching + // neither of the ones before it. + let seeked = reader + .read_at(6 << 20, 64 << 10) + .await + .expect("read from the middle of the file"); + let want = &contents[(6 << 20)..(6 << 20) + (64 << 10)]; + assert_eq!( + seeked, want, + "a mid-file read must return the bytes at that offset, not an earlier block's" + ); + + // The short tail, and a read that runs past EOF: clamped, never padded. + let tail = reader + .read_at(8 << 20, 4 << 20) + .await + .expect("read the short final block"); + assert_eq!( + tail, + &contents[8 << 20..], + "the final block is 1 MiB, and a read past EOF is clamped to it" + ); + assert!( + reader + .read_at(contents.len() as u64, 4096) + .await + .expect("a read at EOF is not an error") + .is_empty(), + "a read starting at EOF yields nothing" + ); + + // `download_range` is the one-shot form of the same thing. + let ranged = visitor + .download_range(&file, 5 << 20, 4096) + .await + .expect("download_range"); + assert_eq!(ranged, &contents[(5 << 20)..(5 << 20) + 4096]); + + // And the whole-file path must still agree with all of it — it now drives a + // paged, concurrent block pipeline rather than one un-paged sequential loop. + let whole = visitor + .download_file(&file) + .await + .expect("visitor download"); + assert_eq!( + whole, contents, + "the whole-file download must match the uploaded bytes" + ); + + cleanup(client, &folder).await; +} + +/// A visitor's node listing must carry the uploader's claimed metadata, which it +/// previously left as `None` — a catalog has nothing to show without a size. +#[tokio::test] +#[ignore = "live: needs test-account credentials"] +async fn a_visitor_sees_claimed_size_and_digest_on_shared_files() { + let Some(live) = common::live_client().await else { + return; + }; + let client = &live.client; + + let folder = scratch_folder(client, "public-xattr").await; + + let contents = positional_payload(64 << 10); + let name = format!("claimed-{}.bin", common::unique_suffix()); + let file = client + .upload_file(&folder, &name, "application/octet-stream", &contents) + .await + .expect("upload_file"); + + let link = client + .create_public_link(&folder, MemberRole::Viewer, None, None) + .await + .expect("create_public_link"); + let url = link.url.clone().expect("a created link must carry its URL"); + + let visitor = ProtonDrivePublicLinkClient::open(common::config(), &url, None) + .await + .expect("open public link"); + + let node = visitor + .get_node(&file) + .await + .expect("visitor node lookup") + .expect("the shared file must exist"); + + match &node.kind { + proton_drive_rs::NodeKind::File { + claimed_size, + content_sha1, + .. + } => { + assert_eq!( + *claimed_size, + Some(contents.len() as i64), + "the visitor must read the uploader's claimed size from the link details XAttr" + ); + assert!( + content_sha1.is_some(), + "and the content digest alongside it" + ); + } + other => panic!("expected a file, got {other:?}"), + } + + cleanup(client, &folder).await; +} + +/// Renewing the session must not strand a reader that was already handed out — +/// the whole reason the session is swapped in place rather than replaced. +/// +/// Simulates the expiry rather than waiting hours for one: an explicit +/// `refresh_session` puts the reader on a session it was not opened with, which +/// is exactly the state a real expiry leaves it in once recovery has run. +#[tokio::test] +#[ignore = "live: needs test-account credentials"] +async fn refreshing_the_session_keeps_an_open_reader_working() { + let Some(live) = common::live_client().await else { + return; + }; + let client = &live.client; + + let folder = scratch_folder(client, "public-renew").await; + + let contents = positional_payload(5 << 20); + let name = format!("renew-{}.bin", common::unique_suffix()); + let file = client + .upload_file(&folder, &name, "application/octet-stream", &contents) + .await + .expect("upload_file"); + + let link = client + .create_public_link(&folder, MemberRole::Viewer, None, None) + .await + .expect("create_public_link"); + let url = link.url.clone().expect("a created link must carry its URL"); + + let visitor = ProtonDrivePublicLinkClient::open(common::config(), &url, None) + .await + .expect("open public link"); + + let reader = visitor.open_revision(&file).await.expect("open_revision"); + + let before = reader.read_at(0, 4096).await.expect("read before refresh"); + assert_eq!(before, &contents[..4096]); + + visitor + .refresh_session() + .await + .expect("refresh_session must replay the handshake"); + + // Past the first block, so this genuinely fetches under the new session + // rather than replaying anything cached. + let after = reader + .read_at(4 << 20, 4096) + .await + .expect("a reader opened before the refresh must keep working"); + assert_eq!(after, &contents[(4 << 20)..(4 << 20) + 4096]); + + cleanup(client, &folder).await; +} + +/// Thumbnails on the visitor path: the same block-storage fetch and content-key +/// decrypt as the authenticated one, over `drive/unauth/`. +#[tokio::test] +#[ignore = "live: needs test-account credentials"] +async fn a_visitor_can_download_thumbnails_of_shared_files() { + let Some(live) = common::live_client().await else { + return; + }; + let client = &live.client; + + let folder = scratch_folder(client, "public-thumb").await; + + let contents = positional_payload(32 << 10); + let thumbnail = b"not a real jpeg, but it round-trips".to_vec(); + let name = format!("thumbed-{}.bin", common::unique_suffix()); + let file = client + .upload_file_from( + &folder, + &name, + "application/octet-stream", + std::io::Cursor::new(contents.clone()), + contents.len() as i64, + vec![ + Thumbnail::new(ThumbnailType::Thumbnail, thumbnail.clone()) + .expect("small thumbnail"), + ], + None, + false, + ) + .await + .expect("upload_file_from with a thumbnail"); + + let link = client + .create_public_link(&folder, MemberRole::Viewer, None, None) + .await + .expect("create_public_link"); + let url = link.url.clone().expect("a created link must carry its URL"); + + let visitor = ProtonDrivePublicLinkClient::open(common::config(), &url, None) + .await + .expect("open public link"); + + let single = visitor + .download_thumbnail(&file, ThumbnailType::Thumbnail) + .await + .expect("visitor thumbnail download"); + assert_eq!( + single.as_deref(), + Some(thumbnail.as_slice()), + "the visitor must decrypt the thumbnail it was shown" + ); + + let batch = visitor + .enumerate_thumbnails(std::slice::from_ref(&file), ThumbnailType::Thumbnail) + .await + .expect("visitor thumbnail batch"); + assert_eq!(batch.len(), 1, "one file in, one result out"); + assert_eq!( + batch[0] + .result + .as_deref() + .expect("the batch entry succeeded"), + thumbnail.as_slice() + ); + + cleanup(client, &folder).await; +} diff --git a/crates/proton-drive-rs/tests/spawnable.rs b/crates/proton-drive-rs/tests/spawnable.rs index e6027f6..acf6f0c 100644 --- a/crates/proton-drive-rs/tests/spawnable.rs +++ b/crates/proton-drive-rs/tests/spawnable.rs @@ -16,7 +16,7 @@ use std::sync::Arc; use tokio::task::JoinHandle; -use proton_drive_rs::{ProtonDriveClient, RevisionReader}; +use proton_drive_rs::{ProtonDriveClient, ProtonDrivePublicLinkClient, RevisionReader}; use proton_sdk::ids::NodeUid; /// `download_file` inside a spawned task. @@ -128,3 +128,49 @@ fn reader_read_at_is_spawnable( Ok::<_, proton_sdk::error::ProtonError>(()) }) } + +/// The same guard on the visitor path, which is the one that actually matters +/// for a streaming player: `open_revision` over a public link, then several +/// concurrent `read_at`s off one shared reader inside spawned tasks. +/// +/// The public client reaches this through a different route than the +/// authenticated one — its transport wraps a renewable session behind a trait +/// object — so `Send`ness has to be re-proven, not assumed. +fn public_link_reader_read_at_is_spawnable( + client: ProtonDrivePublicLinkClient, + uid: NodeUid, +) -> JoinHandle> { + tokio::spawn(async move { + let reader: Arc = Arc::new(client.open_revision(&uid).await?); + let mut set = tokio::task::JoinSet::new(); + for i in 0..4u64 { + let reader = reader.clone(); + set.spawn(async move { reader.read_at(i * 4096, 4096).await }); + } + while let Some(joined) = set.join_next().await { + joined.expect("task panicked")?; + } + Ok::<_, proton_sdk::error::ProtonError>(()) + }) +} + +/// `download_file_to` on the visitor path, which now drives the same `buffered` +/// block pipeline the authenticated one does. +fn public_link_download_file_to_is_spawnable( + client: ProtonDrivePublicLinkClient, + uid: NodeUid, +) -> JoinHandle> { + tokio::spawn(async move { + let mut sink: Vec = Vec::new(); + client.download_file_to(&uid, &mut sink).await + }) +} + +/// `enumerate_nodes` on the visitor path: chunked and fanned out with +/// `buffered` over cloned clients. +fn public_link_enumerate_nodes_is_spawnable( + client: ProtonDrivePublicLinkClient, + uids: Vec, +) -> JoinHandle>> { + tokio::spawn(async move { client.enumerate_nodes(&uids).await }) +} diff --git a/crates/proton-sdk/src/http.rs b/crates/proton-sdk/src/http.rs index 4e3ec55..7607ad1 100644 --- a/crates/proton-sdk/src/http.rs +++ b/crates/proton-sdk/src/http.rs @@ -48,6 +48,16 @@ pub struct ApiHttpClient { /// root. Empty by default. Lives on the outer struct (not `Inner`) so clones /// can carry different prefixes while sharing one token/telemetry store. route_prefix: Arc, + /// Whether a 401 should be met with a token refresh and one retry. + /// + /// True for an ordinary bearer session. False for an anonymous public-link + /// session, whose refresh token is deliberately empty + /// (`public_link.rs::PublicLinkSession::auth`) — there, refreshing posts an + /// empty token to `auth/v4/refresh` and the caller sees an error about + /// *refresh* instead of the 401 that actually happened. Lives on the outer + /// struct alongside `route_prefix`, for the same reason: clones differ + /// while sharing one token store. + refresh_tokens: bool, } /// Callback invoked after a successful token refresh. @@ -98,6 +108,7 @@ impl ApiHttpClient { on_tokens_refreshed: std::sync::Mutex::new(None), }), route_prefix: Arc::from(""), + refresh_tokens: true, }) } @@ -110,6 +121,24 @@ impl ApiHttpClient { Self { inner: Arc::clone(&self.inner), route_prefix: route.into(), + refresh_tokens: self.refresh_tokens, + } + } + + /// Derive a clone that does **not** try to refresh its tokens on a 401, + /// sharing this client's token store, telemetry sink and connection pool. + /// + /// For a session that has no refresh token to spend — an anonymous + /// public-link session is minted with an empty one — refreshing turns a + /// clean "your session is gone" 401 into a confusing failure of + /// `auth/v4/refresh` itself. With refresh off, the 401 reaches the caller + /// intact as `ProtonError::Api { http_status: 401 }`, which is what a + /// re-handshake path needs in order to recognise its cue. + pub fn without_token_refresh(&self) -> Self { + Self { + inner: Arc::clone(&self.inner), + route_prefix: Arc::clone(&self.route_prefix), + refresh_tokens: false, } } @@ -288,7 +317,7 @@ impl ApiHttpClient { let response = self .send_multipart_with_token(path, &metadata, binary_parts, &rejected_token) .await?; - let response = if response.status() == StatusCode::UNAUTHORIZED { + let response = if response.status() == StatusCode::UNAUTHORIZED && self.refresh_tokens { let bytes = response.bytes().await?; if let Ok(envelope) = serde_json::from_slice::(&bytes) && matches!( @@ -349,7 +378,7 @@ impl ApiHttpClient { .send_with_token(method.clone(), path, body, &access_token) .await?; - let response = if response.status() == StatusCode::UNAUTHORIZED { + let response = if response.status() == StatusCode::UNAUTHORIZED && self.refresh_tokens { self.handle_unauthorized(method, path, body, response, access_token) .await? } else { @@ -895,4 +924,107 @@ mod tests { .any(|(k, v)| *k == "status" && v == "200") ); } + + /// A client built with [`ApiHttpClient::without_token_refresh`] hands the + /// 401 straight to the caller instead of spending a refresh token on it. + /// + /// The request count is the real assertion: an ordinary client answers a 401 + /// by posting to `auth/v4/refresh` and retrying, so a second connection + /// would prove the opt-out did not take. A public-link session has no + /// refresh token to spend, and needs the bare 401 to know it must re-run the + /// SRP handshake. + #[tokio::test] + async fn a_client_without_token_refresh_surfaces_the_401() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let requests = Arc::new(AtomicUsize::new(0)); + + // Answers every connection with a 401, so a client that retries after a + // refresh gets counted rather than hanging. + let seen = requests.clone(); + let server = tokio::spawn(async move { + loop { + let (mut sock, _) = listener.accept().await.unwrap(); + seen.fetch_add(1, Ordering::SeqCst); + let mut buf = [0u8; 2048]; + let _ = sock.read(&mut buf).await.unwrap(); + let body = br#"{"Code":401,"Error":"Invalid access token"}"#; + let head = format!( + "HTTP/1.1 401 Unauthorized\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + sock.write_all(head.as_bytes()).await.unwrap(); + sock.write_all(body).await.unwrap(); + sock.flush().await.unwrap(); + } + }); + + let config = ProtonClientConfiguration::new("test@1.0") + .with_base_url(format!("http://{addr}/")) + .with_retry_policy(RetryPolicy::disabled()); + let client = ApiHttpClient::new( + config, + SessionId::from("test-session"), + // Empty, exactly as a public-link session is minted. + Tokens { + access_token: "access".into(), + refresh_token: String::new(), + }, + ) + .unwrap() + .without_token_refresh(); + + let error = client + .get::("some/path") + .await + .expect_err("a 401 is an error"); + + match error { + ProtonError::Api(e) => assert_eq!(e.http_status, 401, "the 401 reaches the caller"), + other => panic!("expected an api error, got {other:?}"), + } + assert_eq!( + requests.load(Ordering::SeqCst), + 1, + "no refresh attempt, no retry" + ); + + server.abort(); + } + + /// The opt-out rides along on `with_base_route`, so a Drive/public-link + /// client derived from a non-refreshing one does not silently regain the + /// refresh behaviour. + #[test] + fn the_token_refresh_opt_out_survives_a_base_route_change() { + let client = ApiHttpClient::new( + ProtonClientConfiguration::new("test@1.0"), + SessionId::from("test-session"), + Tokens { + access_token: "access".into(), + refresh_token: String::new(), + }, + ) + .unwrap(); + + assert!(client.refresh_tokens, "an ordinary session refreshes"); + assert!( + !client.without_token_refresh().refresh_tokens, + "the opt-out takes" + ); + assert!( + !client + .without_token_refresh() + .with_base_route("drive/unauth/") + .refresh_tokens, + "and survives a route change" + ); + assert!( + client.with_base_route("drive/").refresh_tokens, + "an ordinary session keeps refreshing across a route change" + ); + } }