From bffe18c6bd2a0991fb42365aeb435bad3c07effb Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 10 Aug 2026 14:28:32 -0400 Subject: [PATCH 1/4] feat(dgw): zip multi-clip recording downloads Add GET /jet/jrec/pull/{id} that streams a ZIP of recording.json and every clip listed in the session manifest. Callers no longer need to guess a single clip filename when a reconnect produced multiple files. The existing per-file pull route is unchanged for player and granular access. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- Cargo.lock | 62 +++- devolutions-gateway/Cargo.toml | 1 + devolutions-gateway/openapi/doc/index.adoc | 97 +++++ .../openapi/dotnet-client/README.md | 1 + .../openapi/dotnet-client/docs/JrecApi.md | 102 ++++++ .../Devolutions.Gateway.Client/Api/JrecApi.cs | 163 +++++++++ devolutions-gateway/openapi/gateway-api.yaml | 35 ++ .../ts-angular-client/api/jrec.service.ts | 61 ++++ devolutions-gateway/src/api/jrec.rs | 332 +++++++++++++++++- devolutions-gateway/src/openapi.rs | 1 + 10 files changed, 838 insertions(+), 17 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0240888a9..40994c9bd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -383,6 +383,20 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "async_zip" +version = "0.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d8c50d65ce1b0e0cb65a785ff615f78860d7754290647d3b983208daa4f85e6" +dependencies = [ + "crc32fast", + "futures-lite", + "pin-project 1.1.13", + "thiserror 2.0.18", + "tokio 1.52.3", + "tokio-util", +] + [[package]] name = "atomic-polyfill" version = "1.0.3" @@ -1530,7 +1544,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.118", ] [[package]] @@ -1764,6 +1778,7 @@ dependencies = [ "anyhow", "argon2", "async-trait", + "async_zip", "axum 0.8.9", "axum-extra", "backoff", @@ -2296,7 +2311,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2536,6 +2551,19 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" +[[package]] +name = "futures-lite" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "parking", + "pin-project-lite 0.2.17", +] + [[package]] name = "futures-macro" version = "0.3.32" @@ -2612,7 +2640,7 @@ dependencies = [ "libc", "log", "rustversion", - "windows-link 0.1.3", + "windows-link 0.2.1", "windows-result 0.4.1", ] @@ -3119,7 +3147,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite 0.2.17", - "socket2 0.5.10", + "socket2 0.6.5", "system-configuration", "tokio 1.52.3", "tower-service", @@ -3658,7 +3686,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -4809,7 +4837,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5056,6 +5084,12 @@ dependencies = [ "sha2 0.11.0", ] +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -5962,7 +5996,7 @@ dependencies = [ "quinn-udp", "rustc-hash 2.1.3", "rustls 0.23.42", - "socket2 0.5.10", + "socket2 0.6.5", "thiserror 2.0.18", "tokio 1.52.3", "tracing", @@ -6002,9 +6036,9 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2 0.6.5", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6501,7 +6535,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6601,7 +6635,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7129,7 +7163,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7422,7 +7456,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -8864,7 +8898,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/devolutions-gateway/Cargo.toml b/devolutions-gateway/Cargo.toml index b4a5dd66c..9143dd41f 100644 --- a/devolutions-gateway/Cargo.toml +++ b/devolutions-gateway/Cargo.toml @@ -107,6 +107,7 @@ tungstenite = "0.29" # Should be the same version as `axum` (we perform error do tokio-tungstenite = { version = "0.29", features = ["rustls-tls-native-roots"] } # Should use the same version of tungstenite as `axum` http-body-util = "0.1" tokio-retry = "0.3" +async_zip = { version = "0.0.18", default-features = false, features = ["tokio"] } # OpenAPI generator utoipa = { version = "4.2", default-features = false, features = ["uuid", "time"], optional = true } diff --git a/devolutions-gateway/openapi/doc/index.adoc b/devolutions-gateway/openapi/doc/index.adoc index 7bc10742c..8ccf63f32 100644 --- a/devolutions-gateway/openapi/doc/index.adoc +++ b/devolutions-gateway/openapi/doc/index.adoc @@ -938,6 +938,103 @@ ifdef::internal-generation[] endif::internal-generation[] +[.pullRecordingSession] +==== pullRecordingSession + +`GET /jet/jrec/pull/{id}` + +Downloads an entire recorded session as a ZIP archive + +===== Description + +The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + + +// markup not found, no include::{specDir}jet/jrec/pull/\{id\}/GET/spec.adoc[opts=optional] + + + +===== Security + +[cols="2,1,1"] +|=== +| Name | Type | Scheme + +| `jrec_token` +| http +| bearer +|=== + +===== Parameters + +====== Path Parameters + +[cols="2,3,1,1,1"] +|=== +|Name| Description| Required| Default| Pattern + +| id +| Recorded session ID +| X +| null +| + +|=== + + + + + + +===== Return Type + + +<> + + +===== Content Type + +* application/zip + +===== Responses + +.HTTP Response Codes +[cols="2,3,1"] +|=== +| Code | Message | Datatype + + +| 200 +| ZIP archive containing the recording session +| <> + + +| 401 +| Invalid or missing authorization token +| <<>> + + +| 403 +| Insufficient permissions +| <<>> + + +| 404 +| Recording not found +| <<>> + +|=== + + +ifdef::internal-generation[] +===== Implementation + +// markup not found, no include::{specDir}jet/jrec/pull/\{id\}/GET/implementation.adoc[opts=optional] + + +endif::internal-generation[] + + [.Jrl] === Jrl diff --git a/devolutions-gateway/openapi/dotnet-client/README.md b/devolutions-gateway/openapi/dotnet-client/README.md index 38bc4d138..c7b5aafc6 100644 --- a/devolutions-gateway/openapi/dotnet-client/README.md +++ b/devolutions-gateway/openapi/dotnet-client/README.md @@ -151,6 +151,7 @@ Class | Method | HTTP request | Description *JrecApi* | [**DeleteRecording**](docs/JrecApi.md#deleterecording) | **DELETE** /jet/jrec/delete/{id} | Deletes a recording stored on this instance *JrecApi* | [**ListRecordings**](docs/JrecApi.md#listrecordings) | **GET** /jet/jrec/list | Lists all recordings stored on this instance *JrecApi* | [**PullRecordingFile**](docs/JrecApi.md#pullrecordingfile) | **GET** /jet/jrec/pull/{id}/{filename} | Retrieves a recording file for a given session +*JrecApi* | [**PullRecordingSession**](docs/JrecApi.md#pullrecordingsession) | **GET** /jet/jrec/pull/{id} | Downloads an entire recorded session as a ZIP archive *JrlApi* | [**GetJrlInfo**](docs/JrlApi.md#getjrlinfo) | **GET** /jet/jrl/info | Retrieves current JRL (Json Revocation List) info *JrlApi* | [**UpdateJrl**](docs/JrlApi.md#updatejrl) | **POST** /jet/jrl | Updates JRL (Json Revocation List) using a JRL token *NetApi* | [**GetNetConfig**](docs/NetApi.md#getnetconfig) | **GET** /jet/net/config | Lists network interfaces diff --git a/devolutions-gateway/openapi/dotnet-client/docs/JrecApi.md b/devolutions-gateway/openapi/dotnet-client/docs/JrecApi.md index 6b6f3e7a1..bcdc392d0 100644 --- a/devolutions-gateway/openapi/dotnet-client/docs/JrecApi.md +++ b/devolutions-gateway/openapi/dotnet-client/docs/JrecApi.md @@ -8,6 +8,7 @@ All URIs are relative to *http://localhost* | [**DeleteRecording**](JrecApi.md#deleterecording) | **DELETE** /jet/jrec/delete/{id} | Deletes a recording stored on this instance | | [**ListRecordings**](JrecApi.md#listrecordings) | **GET** /jet/jrec/list | Lists all recordings stored on this instance | | [**PullRecordingFile**](JrecApi.md#pullrecordingfile) | **GET** /jet/jrec/pull/{id}/{filename} | Retrieves a recording file for a given session | +| [**PullRecordingSession**](JrecApi.md#pullrecordingsession) | **GET** /jet/jrec/pull/{id} | Downloads an entire recorded session as a ZIP archive | # **DeleteManyRecordings** @@ -410,3 +411,104 @@ catch (ApiException e) [[Back to top]](#) [[Back to API list]](../README.md#documentation-for-api-endpoints) [[Back to Model list]](../README.md#documentation-for-models) [[Back to README]](../README.md) + +# **PullRecordingSession** +> FileParameter PullRecordingSession (Guid id) + +Downloads an entire recorded session as a ZIP archive + +The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + +### Example +```csharp +using System.Collections.Generic; +using System.Diagnostics; +using System.Net.Http; +using Devolutions.Gateway.Client.Api; +using Devolutions.Gateway.Client.Client; +using Devolutions.Gateway.Client.Model; + +namespace Example +{ + public class PullRecordingSessionExample + { + public static void Main() + { + Configuration config = new Configuration(); + config.BasePath = "http://localhost"; + // Configure Bearer token for authorization: jrec_token + config.AccessToken = "YOUR_BEARER_TOKEN"; + + // create instances of HttpClient, HttpClientHandler to be reused later with different Api classes + HttpClient httpClient = new HttpClient(); + HttpClientHandler httpClientHandler = new HttpClientHandler(); + var apiInstance = new JrecApi(httpClient, config, httpClientHandler); + var id = "id_example"; // Guid | Recorded session ID + + try + { + // Downloads an entire recorded session as a ZIP archive + FileParameter result = apiInstance.PullRecordingSession(id); + Debug.WriteLine(result); + } + catch (ApiException e) + { + Debug.Print("Exception when calling JrecApi.PullRecordingSession: " + e.Message); + Debug.Print("Status Code: " + e.ErrorCode); + Debug.Print(e.StackTrace); + } + } + } +} +``` + +#### Using the PullRecordingSessionWithHttpInfo variant +This returns an ApiResponse object which contains the response data, status code and headers. + +```csharp +try +{ + // Downloads an entire recorded session as a ZIP archive + ApiResponse response = apiInstance.PullRecordingSessionWithHttpInfo(id); + Debug.Write("Status Code: " + response.StatusCode); + Debug.Write("Response Headers: " + response.Headers); + Debug.Write("Response Body: " + response.Data); +} +catch (ApiException e) +{ + Debug.Print("Exception when calling JrecApi.PullRecordingSessionWithHttpInfo: " + e.Message); + Debug.Print("Status Code: " + e.ErrorCode); + Debug.Print(e.StackTrace); +} +``` + +### Parameters + +| Name | Type | Description | Notes | +|------|------|-------------|-------| +| **id** | **Guid** | Recorded session ID | | + +### Return type + +[**FileParameter**](FileParameter.md) + +### Authorization + +[jrec_token](../README.md#jrec_token) + +### HTTP request headers + + - **Content-Type**: Not defined + - **Accept**: application/zip + + +### HTTP response details +| Status code | Description | Response headers | +|-------------|-------------|------------------| +| **200** | ZIP archive containing the recording session | - | +| **401** | Invalid or missing authorization token | - | +| **403** | Insufficient permissions | - | +| **404** | Recording not found | - | + +[[Back to top]](#) [[Back to API list]](../README.md#documentation-for-api-endpoints) [[Back to Model list]](../README.md#documentation-for-models) [[Back to README]](../README.md) + diff --git a/devolutions-gateway/openapi/dotnet-client/src/Devolutions.Gateway.Client/Api/JrecApi.cs b/devolutions-gateway/openapi/dotnet-client/src/Devolutions.Gateway.Client/Api/JrecApi.cs index dd2fe3342..17a031fcd 100644 --- a/devolutions-gateway/openapi/dotnet-client/src/Devolutions.Gateway.Client/Api/JrecApi.cs +++ b/devolutions-gateway/openapi/dotnet-client/src/Devolutions.Gateway.Client/Api/JrecApi.cs @@ -105,6 +105,27 @@ public interface IJrecApiSync : IApiAccessor /// Name of recording file to retrieve /// ApiResponse of FileParameter ApiResponse PullRecordingFileWithHttpInfo(Guid id, string filename); + /// + /// Downloads an entire recorded session as a ZIP archive + /// + /// + /// The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + /// + /// Thrown when fails to make API call + /// Recorded session ID + /// FileParameter + FileParameter PullRecordingSession(Guid id); + + /// + /// Downloads an entire recorded session as a ZIP archive + /// + /// + /// The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + /// + /// Thrown when fails to make API call + /// Recorded session ID + /// ApiResponse of FileParameter + ApiResponse PullRecordingSessionWithHttpInfo(Guid id); #endregion Synchronous Operations } @@ -208,6 +229,29 @@ public interface IJrecApiAsync : IApiAccessor /// Cancellation Token to cancel the request. /// Task of ApiResponse (FileParameter) System.Threading.Tasks.Task> PullRecordingFileWithHttpInfoAsync(Guid id, string filename, System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)); + /// + /// Downloads an entire recorded session as a ZIP archive + /// + /// + /// The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + /// + /// Thrown when fails to make API call + /// Recorded session ID + /// Cancellation Token to cancel the request. + /// Task of FileParameter + System.Threading.Tasks.Task PullRecordingSessionAsync(Guid id, System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)); + + /// + /// Downloads an entire recorded session as a ZIP archive + /// + /// + /// The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + /// + /// Thrown when fails to make API call + /// Recorded session ID + /// Cancellation Token to cancel the request. + /// Task of ApiResponse (FileParameter) + System.Threading.Tasks.Task> PullRecordingSessionWithHttpInfoAsync(Guid id, System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)); #endregion Asynchronous Operations } @@ -917,5 +961,124 @@ public Devolutions.Gateway.Client.Client.ApiResponse PullRecordin return localVarResponse; } + /// + /// Downloads an entire recorded session as a ZIP archive The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + /// + /// Thrown when fails to make API call + /// Recorded session ID + /// FileParameter + public FileParameter PullRecordingSession(Guid id) + { + Devolutions.Gateway.Client.Client.ApiResponse localVarResponse = PullRecordingSessionWithHttpInfo(id); + return localVarResponse.Data; + } + + /// + /// Downloads an entire recorded session as a ZIP archive The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + /// + /// Thrown when fails to make API call + /// Recorded session ID + /// ApiResponse of FileParameter + public Devolutions.Gateway.Client.Client.ApiResponse PullRecordingSessionWithHttpInfo(Guid id) + { + Devolutions.Gateway.Client.Client.RequestOptions localVarRequestOptions = new Devolutions.Gateway.Client.Client.RequestOptions(); + + string[] _contentTypes = new string[] { + }; + + // to determine the Accept header + string[] _accepts = new string[] { + "application/zip" + }; + + var localVarContentType = Devolutions.Gateway.Client.Client.ClientUtils.SelectHeaderContentType(_contentTypes); + if (localVarContentType != null) localVarRequestOptions.HeaderParameters.Add("Content-Type", localVarContentType); + + var localVarAccept = Devolutions.Gateway.Client.Client.ClientUtils.SelectHeaderAccept(_accepts); + if (localVarAccept != null) localVarRequestOptions.HeaderParameters.Add("Accept", localVarAccept); + + localVarRequestOptions.PathParameters.Add("id", Devolutions.Gateway.Client.Client.ClientUtils.ParameterToString(id)); // path parameter + + // authentication (jrec_token) required + // bearer authentication required + if (!string.IsNullOrEmpty(this.Configuration.AccessToken) && !localVarRequestOptions.HeaderParameters.ContainsKey("Authorization")) + { + localVarRequestOptions.HeaderParameters.Add("Authorization", "Bearer " + this.Configuration.AccessToken); + } + + // make the HTTP request + var localVarResponse = this.Client.Get("/jet/jrec/pull/{id}", localVarRequestOptions, this.Configuration); + + if (this.ExceptionFactory != null) + { + Exception _exception = this.ExceptionFactory("PullRecordingSession", localVarResponse); + if (_exception != null) throw _exception; + } + + return localVarResponse; + } + + /// + /// Downloads an entire recorded session as a ZIP archive The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + /// + /// Thrown when fails to make API call + /// Recorded session ID + /// Cancellation Token to cancel the request. + /// Task of FileParameter + public async System.Threading.Tasks.Task PullRecordingSessionAsync(Guid id, System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) + { + Devolutions.Gateway.Client.Client.ApiResponse localVarResponse = await PullRecordingSessionWithHttpInfoAsync(id, cancellationToken).ConfigureAwait(false); + return localVarResponse.Data; + } + + /// + /// Downloads an entire recorded session as a ZIP archive The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + /// + /// Thrown when fails to make API call + /// Recorded session ID + /// Cancellation Token to cancel the request. + /// Task of ApiResponse (FileParameter) + public async System.Threading.Tasks.Task> PullRecordingSessionWithHttpInfoAsync(Guid id, System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) + { + + Devolutions.Gateway.Client.Client.RequestOptions localVarRequestOptions = new Devolutions.Gateway.Client.Client.RequestOptions(); + + string[] _contentTypes = new string[] { + }; + + // to determine the Accept header + string[] _accepts = new string[] { + "application/zip" + }; + + + var localVarContentType = Devolutions.Gateway.Client.Client.ClientUtils.SelectHeaderContentType(_contentTypes); + if (localVarContentType != null) localVarRequestOptions.HeaderParameters.Add("Content-Type", localVarContentType); + + var localVarAccept = Devolutions.Gateway.Client.Client.ClientUtils.SelectHeaderAccept(_accepts); + if (localVarAccept != null) localVarRequestOptions.HeaderParameters.Add("Accept", localVarAccept); + + localVarRequestOptions.PathParameters.Add("id", Devolutions.Gateway.Client.Client.ClientUtils.ParameterToString(id)); // path parameter + + // authentication (jrec_token) required + // bearer authentication required + if (!string.IsNullOrEmpty(this.Configuration.AccessToken) && !localVarRequestOptions.HeaderParameters.ContainsKey("Authorization")) + { + localVarRequestOptions.HeaderParameters.Add("Authorization", "Bearer " + this.Configuration.AccessToken); + } + + // make the HTTP request + + var localVarResponse = await this.AsynchronousClient.GetAsync("/jet/jrec/pull/{id}", localVarRequestOptions, this.Configuration, cancellationToken).ConfigureAwait(false); + + if (this.ExceptionFactory != null) + { + Exception _exception = this.ExceptionFactory("PullRecordingSession", localVarResponse); + if (_exception != null) throw _exception; + } + + return localVarResponse; + } + } } diff --git a/devolutions-gateway/openapi/gateway-api.yaml b/devolutions-gateway/openapi/gateway-api.yaml index c64ef1c77..db493241b 100644 --- a/devolutions-gateway/openapi/gateway-api.yaml +++ b/devolutions-gateway/openapi/gateway-api.yaml @@ -255,6 +255,41 @@ paths: security: - scope_token: - gateway.recordings.read + /jet/jrec/pull/{id}: + get: + tags: + - Jrec + summary: Downloads an entire recorded session as a ZIP archive + description: |- + The archive always contains `recording.json` and every clip listed in that + manifest that is present on disk. A single-clip session is still returned as a ZIP + so callers can use one download contract for every recording. + operationId: PullRecordingSession + parameters: + - name: id + in: path + description: Recorded session ID + required: true + schema: + type: string + format: uuid + responses: + '200': + description: ZIP archive containing the recording session + content: + application/zip: + schema: + type: string + format: binary + '401': + description: Invalid or missing authorization token + '403': + description: Insufficient permissions + '404': + description: Recording not found + security: + - jrec_token: + - pull /jet/jrec/pull/{id}/{filename}: get: tags: diff --git a/devolutions-gateway/openapi/ts-angular-client/api/jrec.service.ts b/devolutions-gateway/openapi/ts-angular-client/api/jrec.service.ts index 208f6898b..b1e8e0a18 100644 --- a/devolutions-gateway/openapi/ts-angular-client/api/jrec.service.ts +++ b/devolutions-gateway/openapi/ts-angular-client/api/jrec.service.ts @@ -384,4 +384,65 @@ export class JrecService { ); } + /** + * Downloads an entire recorded session as a ZIP archive + * The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. + * @param id Recorded session ID + * @param observe set whether or not to return the data Observable as the body, response or events. defaults to returning the body. + * @param reportProgress flag to report request and response progress. + */ + public pullRecordingSession(id: string, observe?: 'body', reportProgress?: boolean, options?: {httpHeaderAccept?: 'application/zip', context?: HttpContext, transferCache?: boolean}): Observable; + public pullRecordingSession(id: string, observe?: 'response', reportProgress?: boolean, options?: {httpHeaderAccept?: 'application/zip', context?: HttpContext, transferCache?: boolean}): Observable>; + public pullRecordingSession(id: string, observe?: 'events', reportProgress?: boolean, options?: {httpHeaderAccept?: 'application/zip', context?: HttpContext, transferCache?: boolean}): Observable>; + public pullRecordingSession(id: string, observe: any = 'body', reportProgress: boolean = false, options?: {httpHeaderAccept?: 'application/zip', context?: HttpContext, transferCache?: boolean}): Observable { + if (id === null || id === undefined) { + throw new Error('Required parameter id was null or undefined when calling pullRecordingSession.'); + } + + let localVarHeaders = this.defaultHeaders; + + let localVarCredential: string | undefined; + // authentication (jrec_token) required + localVarCredential = this.configuration.lookupCredential('jrec_token'); + if (localVarCredential) { + localVarHeaders = localVarHeaders.set('Authorization', 'Bearer ' + localVarCredential); + } + + let localVarHttpHeaderAcceptSelected: string | undefined = options && options.httpHeaderAccept; + if (localVarHttpHeaderAcceptSelected === undefined) { + // to determine the Accept header + const httpHeaderAccepts: string[] = [ + 'application/zip' + ]; + localVarHttpHeaderAcceptSelected = this.configuration.selectHeaderAccept(httpHeaderAccepts); + } + if (localVarHttpHeaderAcceptSelected !== undefined) { + localVarHeaders = localVarHeaders.set('Accept', localVarHttpHeaderAcceptSelected); + } + + let localVarHttpContext: HttpContext | undefined = options && options.context; + if (localVarHttpContext === undefined) { + localVarHttpContext = new HttpContext(); + } + + let localVarTransferCache: boolean | undefined = options && options.transferCache; + if (localVarTransferCache === undefined) { + localVarTransferCache = true; + } + + + let localVarPath = `/jet/jrec/pull/${this.configuration.encodeParam({name: "id", value: id, in: "path", style: "simple", explode: false, dataType: "string", dataFormat: "uuid"})}`; + return this.httpClient.request('get', `${this.configuration.basePath}${localVarPath}`, + { + context: localVarHttpContext, + responseType: "blob", + withCredentials: this.configuration.withCredentials, + headers: localVarHeaders, + observe: observe, + transferCache: localVarTransferCache, + reportProgress: reportProgress + } + ); + } + } diff --git a/devolutions-gateway/src/api/jrec.rs b/devolutions-gateway/src/api/jrec.rs index f62597bfd..9dc443e6c 100644 --- a/devolutions-gateway/src/api/jrec.rs +++ b/devolutions-gateway/src/api/jrec.rs @@ -4,16 +4,20 @@ use std::path::Path; use std::time::Duration; use anyhow::Context as _; +use async_zip::tokio::write::ZipFileWriter; +use async_zip::{Compression, ZipEntryBuilder}; use axum::extract::ws::{CloseFrame, WebSocket}; use axum::extract::{self, ConnectInfo, Query, State, WebSocketUpgrade}; -use axum::http::header::{CONTENT_TYPE, HeaderValue}; -use axum::response::Response; +use axum::http::header::{CONTENT_DISPOSITION, CONTENT_TYPE, HeaderValue}; +use axum::response::{IntoResponse as _, Response}; use axum::routing::{delete, get}; use axum::{Json, Router}; +use axum_extra::body::AsyncReadBody; use cadeau::xmf; use camino::{Utf8Path, Utf8PathBuf}; use devolutions_gateway_task::ShutdownSignal; use hyper::StatusCode; +use tokio::io::AsyncWrite; use tracing::Instrument as _; use uuid::Uuid; @@ -30,6 +34,7 @@ pub fn make_router(state: DgwState) -> Router { .route("/delete/{id}", delete(jrec_delete)) .route("/delete", delete(jrec_delete_many)) .route("/list", get(list_recordings)) + .route("/pull/{id}", get(pull_recording_session)) .route("/pull/{id}/{filename}", get(pull_recording_file)) .route("/play", get(get_player)) .route("/play/{*path}", get(get_player)) @@ -464,6 +469,80 @@ pub(crate) async fn list_recordings( } } +/// Downloads an entire recorded session as a ZIP archive +/// +/// The archive always contains `recording.json` and every clip listed in that +/// manifest that is present on disk. A single-clip session is still returned as a ZIP +/// so callers can use one download contract for every recording. +#[cfg_attr(feature = "openapi", utoipa::path( + get, + operation_id = "PullRecordingSession", + tag = "Jrec", + path = "/jet/jrec/pull/{id}", + params( + ("id" = Uuid, Path, description = "Recorded session ID"), + ), + responses( + (status = 200, description = "ZIP archive containing the recording session", body = Vec, content_type = "application/zip"), + (status = 401, description = "Invalid or missing authorization token"), + (status = 403, description = "Insufficient permissions"), + (status = 404, description = "Recording not found"), + ), + security(("jrec_token" = ["pull"])), +))] +pub(crate) async fn pull_recording_session( + State(DgwState { conf_handle, .. }): State, + extract::Path(id): extract::Path, + JrecToken(claims): JrecToken, +) -> Result { + if id != claims.jet_aid { + return Err(HttpError::forbidden().msg("not allowed to read this recording")); + } + + let recording_dir = conf_handle.get_conf().recording_path.join(id.to_string()); + + if !recording_dir.exists() || !recording_dir.is_dir() { + return Err(HttpError::not_found().msg("requested recording does not exist")); + } + + let entries = list_recording_zip_entries(&recording_dir).await.map_err( + HttpError::internal() + .with_msg("failed to read recording manifest") + .err(), + )?; + + if entries.is_empty() { + return Err(HttpError::not_found().msg("requested recording does not exist")); + } + + // Bounded buffer: backpressure when the client is slower than zip production. + let (zip_writer, zip_reader) = tokio::io::duplex(64 * 1024); + + tokio::spawn(async move { + if let Err(error) = write_recording_zip(zip_writer, &recording_dir, &entries).await { + warn!( + error = format!("{error:#}"), + session.id = %id, + "Failed to stream recording ZIP archive" + ); + } + }); + + let mut response = AsyncReadBody::new(zip_reader).into_response(); + + response + .headers_mut() + .insert(CONTENT_TYPE, HeaderValue::from_static("application/zip")); + + let disposition = format!("attachment; filename=\"{id}.zip\""); + response.headers_mut().insert( + CONTENT_DISPOSITION, + HeaderValue::from_str(&disposition).map_err(HttpError::internal().err())?, + ); + + Ok(response) +} + /// Retrieves a recording file for a given session #[cfg_attr(feature = "openapi", utoipa::path( get, @@ -494,7 +573,7 @@ where { use tower::ServiceExt as _; - if filename.contains("..") || filename.contains('/') || filename.contains('\\') { + if !is_safe_recording_file_name(&filename) { return Err(HttpError::bad_request().msg("invalid file name")); } @@ -531,6 +610,132 @@ where Ok(response) } +/// Minimal `recording.json` view used when packaging a session download. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct RecordingZipManifest { + files: Vec, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct RecordingZipManifestFile { + file_name: String, +} + +fn is_safe_recording_file_name(file_name: &str) -> bool { + !file_name.is_empty() && !file_name.contains("..") && !file_name.contains('/') && !file_name.contains('\\') +} + +/// Builds the ordered list of files to put in a session ZIP. +/// +/// Always starts with `recording.json`, then appends every manifest entry that +/// exists on disk and uses a safe relative file name. +async fn list_recording_zip_entries(recording_dir: &Utf8Path) -> anyhow::Result> { + let manifest_path = recording_dir.join("recording.json"); + if !manifest_path.is_file() { + anyhow::bail!("recording manifest not found"); + } + + let manifest_json = tokio::fs::read(&manifest_path) + .await + .with_context(|| format!("read recording manifest at {manifest_path}"))?; + let manifest: RecordingZipManifest = serde_json::from_slice(&manifest_json).context("parse recording manifest")?; + + let mut entries = Vec::with_capacity(manifest.files.len() + 1); + entries.push("recording.json".to_owned()); + + for file in manifest.files { + if !is_safe_recording_file_name(&file.file_name) { + warn!( + file_name = %file.file_name, + "Skipping unsafe recording file name from manifest" + ); + continue; + } + + let path = recording_dir.join(&file.file_name); + if path.is_file() { + entries.push(file.file_name); + } else { + warn!( + file_name = %file.file_name, + path = %path, + "Skipping missing recording file listed in manifest" + ); + } + } + + Ok(entries) +} + +/// Streams a ZIP archive for the given recording files into `writer`. +/// +/// Entries use the STORED method: recording payloads are already compressed (WebM, etc.), +/// so deflate would mainly burn CPU for little size gain. +async fn write_recording_zip(writer: W, recording_dir: &Utf8Path, entries: &[String]) -> anyhow::Result<()> +where + W: AsyncWrite + Unpin, +{ + use futures::AsyncWriteExt as _; + use tokio::io::AsyncReadExt as _; + + let mut zip_writer = ZipFileWriter::with_tokio(writer); + // Reused across entries to avoid per-file allocation for large multi-clip downloads. + let mut buffer = vec![0u8; 64 * 1024]; + + for file_name in entries { + let path = recording_dir.join(file_name); + let mut file = tokio::fs::File::open(&path) + .await + .with_context(|| format!("open recording file at {path}"))?; + + let builder = ZipEntryBuilder::new(file_name.clone().into(), Compression::Stored); + let mut entry_writer = zip_writer + .write_entry_stream(builder) + .await + .with_context(|| format!("start ZIP entry for {file_name}"))?; + + loop { + let n = file + .read(&mut buffer) + .await + .with_context(|| format!("read recording file at {path}"))?; + if n == 0 { + break; + } + + entry_writer + .write_all(&buffer[..n]) + .await + .with_context(|| format!("write ZIP entry for {file_name}"))?; + } + + entry_writer + .close() + .await + .with_context(|| format!("finish ZIP entry for {file_name}"))?; + } + + zip_writer.close().await.context("finish ZIP archive")?; + Ok(()) +} + +/// Collects a streamed ZIP into memory (tests / small fixtures only). +#[cfg(test)] +async fn collect_recording_zip(recording_dir: &Utf8Path, entries: &[String]) -> anyhow::Result> { + let (writer, mut reader) = tokio::io::duplex(64 * 1024); + let dir = recording_dir.to_owned(); + let entries = entries.to_owned(); + + let writer_task = tokio::spawn(async move { write_recording_zip(writer, &dir, &entries).await }); + + let mut bytes = Vec::new(); + tokio::io::copy(&mut reader, &mut bytes).await?; + writer_task.await.context("zip writer task join")??; + Ok(bytes) +} + async fn get_player( State(DgwState { conf_handle, .. }): State, path: Option>, @@ -612,3 +817,124 @@ async fn shadow_recording( })) } } + +#[cfg(test)] +mod tests { + use async_zip::base::read::mem::ZipFileReader; + + use super::*; + + #[test] + fn rejects_unsafe_recording_file_names() { + assert!(is_safe_recording_file_name("recording-0.webm")); + assert!(is_safe_recording_file_name("recording.json")); + assert!(!is_safe_recording_file_name("")); + assert!(!is_safe_recording_file_name("../secret.webm")); + assert!(!is_safe_recording_file_name("a/b.webm")); + assert!(!is_safe_recording_file_name("a\\b.webm")); + } + + #[tokio::test] + async fn lists_manifest_files_for_zip() { + let dir = tempfile::tempdir().expect("temp dir"); + let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); + + let manifest = serde_json::json!({ + "sessionId": "11111111-1111-1111-1111-111111111111", + "startTime": 1, + "duration": 10, + "files": [ + { "fileName": "recording-0.webm", "startTime": 1, "duration": 5 }, + { "fileName": "recording-1.webm", "startTime": 6, "duration": 5 }, + { "fileName": "missing.webm", "startTime": 11, "duration": 1 }, + { "fileName": "../escape.webm", "startTime": 12, "duration": 1 } + ] + }); + + tokio::fs::write(dir_path.join("recording.json"), manifest.to_string()) + .await + .expect("write manifest"); + tokio::fs::write(dir_path.join("recording-0.webm"), b"clip-zero") + .await + .expect("write clip 0"); + tokio::fs::write(dir_path.join("recording-1.webm"), b"clip-one") + .await + .expect("write clip 1"); + + let entries = list_recording_zip_entries(&dir_path).await.expect("list entries"); + assert_eq!( + entries, + vec![ + "recording.json".to_owned(), + "recording-0.webm".to_owned(), + "recording-1.webm".to_owned(), + ] + ); + } + + #[tokio::test] + async fn streams_zip_with_all_listed_clips() { + let dir = tempfile::tempdir().expect("temp dir"); + let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); + + let manifest = serde_json::json!({ + "sessionId": "22222222-2222-2222-2222-222222222222", + "startTime": 1, + "duration": 4, + "files": [ + { "fileName": "recording-0.webm", "startTime": 1, "duration": 2 }, + { "fileName": "recording-1.webm", "startTime": 3, "duration": 2 } + ] + }); + let manifest_bytes = manifest.to_string().into_bytes(); + + tokio::fs::write(dir_path.join("recording.json"), &manifest_bytes) + .await + .expect("write manifest"); + tokio::fs::write(dir_path.join("recording-0.webm"), b"first-clip") + .await + .expect("write clip 0"); + tokio::fs::write(dir_path.join("recording-1.webm"), b"second-clip") + .await + .expect("write clip 1"); + + let entries = list_recording_zip_entries(&dir_path).await.expect("list entries"); + let zip_bytes = collect_recording_zip(&dir_path, &entries).await.expect("build zip"); + + assert_eq!(&zip_bytes[..2], b"PK"); + + let reader = ZipFileReader::new(zip_bytes).await.expect("parse zip"); + let names: Vec<_> = reader + .file() + .entries() + .iter() + .map(|entry| entry.filename().as_str().expect("utf8 name").to_owned()) + .collect(); + + assert_eq!( + names, + vec![ + "recording.json".to_owned(), + "recording-0.webm".to_owned(), + "recording-1.webm".to_owned(), + ] + ); + + for (index, expected) in [ + manifest_bytes.as_slice(), + b"first-clip".as_slice(), + b"second-clip".as_slice(), + ] + .into_iter() + .enumerate() + { + let mut entry_reader = reader.reader_with_entry(index).await.expect("entry reader"); + let mut content = Vec::new(); + entry_reader + .read_to_end_checked(&mut content) + .await + .expect("read entry"); + assert_eq!(content, expected); + } + } +} diff --git a/devolutions-gateway/src/openapi.rs b/devolutions-gateway/src/openapi.rs index 1e8b3e212..27d8ffdf4 100644 --- a/devolutions-gateway/src/openapi.rs +++ b/devolutions-gateway/src/openapi.rs @@ -22,6 +22,7 @@ use crate::config::dto::{DataEncoding, PubKeyFormat, Subscriber}; crate::api::jrec::jrec_delete, crate::api::jrec::jrec_delete_many, crate::api::jrec::list_recordings, + crate::api::jrec::pull_recording_session, crate::api::jrec::pull_recording_file, crate::api::webapp::sign_app_token, crate::api::webapp::sign_session_token, From 50a3b652aec78bd539aaecf964286c4548b1a92a Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 10 Aug 2026 14:51:54 -0400 Subject: [PATCH 2/4] fix(dgw): harden session recording ZIP download Map missing or corrupt manifests to 404, require pull operation tokens, fail the HTTP body on mid-stream packaging errors, and stop ZIP work on gateway shutdown so clients do not treat truncated archives as success. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- devolutions-gateway/openapi/gateway-api.yaml | 5 +- devolutions-gateway/src/api/jrec.rs | 288 +++++++++++++++---- 2 files changed, 238 insertions(+), 55 deletions(-) diff --git a/devolutions-gateway/openapi/gateway-api.yaml b/devolutions-gateway/openapi/gateway-api.yaml index db493241b..4159240cf 100644 --- a/devolutions-gateway/openapi/gateway-api.yaml +++ b/devolutions-gateway/openapi/gateway-api.yaml @@ -261,9 +261,8 @@ paths: - Jrec summary: Downloads an entire recorded session as a ZIP archive description: |- - The archive always contains `recording.json` and every clip listed in that - manifest that is present on disk. A single-clip session is still returned as a ZIP - so callers can use one download contract for every recording. + The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. + A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. operationId: PullRecordingSession parameters: - name: id diff --git a/devolutions-gateway/src/api/jrec.rs b/devolutions-gateway/src/api/jrec.rs index 9dc443e6c..2c8424705 100644 --- a/devolutions-gateway/src/api/jrec.rs +++ b/devolutions-gateway/src/api/jrec.rs @@ -1,23 +1,26 @@ -use std::fs; use std::net::SocketAddr; use std::path::Path; use std::time::Duration; +use std::{fs, io}; use anyhow::Context as _; use async_zip::tokio::write::ZipFileWriter; use async_zip::{Compression, ZipEntryBuilder}; +use axum::body::Body; use axum::extract::ws::{CloseFrame, WebSocket}; use axum::extract::{self, ConnectInfo, Query, State, WebSocketUpgrade}; use axum::http::header::{CONTENT_DISPOSITION, CONTENT_TYPE, HeaderValue}; -use axum::response::{IntoResponse as _, Response}; +use axum::response::Response; use axum::routing::{delete, get}; use axum::{Json, Router}; -use axum_extra::body::AsyncReadBody; +use bytes::Bytes; use cadeau::xmf; use camino::{Utf8Path, Utf8PathBuf}; use devolutions_gateway_task::ShutdownSignal; +use futures::stream; use hyper::StatusCode; -use tokio::io::AsyncWrite; +use tokio::io::{AsyncReadExt as _, DuplexStream}; +use tokio::sync::oneshot; use tracing::Instrument as _; use uuid::Uuid; @@ -28,6 +31,9 @@ use crate::http::{HttpError, HttpErrorBuilder}; use crate::recording::{PushOutcome, RecordingMessageSender}; use crate::token::{JrecTokenClaims, RecordingFileType, RecordingOperation}; +/// Read/write chunk size for session ZIP streaming (duplex buffer and file copy). +const ZIP_CHUNK_SIZE: usize = 64 * 1024; + pub fn make_router(state: DgwState) -> Router { Router::new() .route("/push/{id}", get(jrec_push)) @@ -471,9 +477,8 @@ pub(crate) async fn list_recordings( /// Downloads an entire recorded session as a ZIP archive /// -/// The archive always contains `recording.json` and every clip listed in that -/// manifest that is present on disk. A single-clip session is still returned as a ZIP -/// so callers can use one download contract for every recording. +/// The archive always contains `recording.json` and every clip listed in that manifest that is present on disk. +/// A single-clip session is still returned as a ZIP so callers can use one download contract for every recording. #[cfg_attr(feature = "openapi", utoipa::path( get, operation_id = "PullRecordingSession", @@ -491,44 +496,42 @@ pub(crate) async fn list_recordings( security(("jrec_token" = ["pull"])), ))] pub(crate) async fn pull_recording_session( - State(DgwState { conf_handle, .. }): State, + State(DgwState { + conf_handle, + shutdown_signal, + .. + }): State, extract::Path(id): extract::Path, JrecToken(claims): JrecToken, ) -> Result { + if claims.jet_rop != RecordingOperation::Pull { + return Err(HttpError::forbidden().msg("expected pull operation")); + } + if id != claims.jet_aid { return Err(HttpError::forbidden().msg("not allowed to read this recording")); } let recording_dir = conf_handle.get_conf().recording_path.join(id.to_string()); - if !recording_dir.exists() || !recording_dir.is_dir() { - return Err(HttpError::not_found().msg("requested recording does not exist")); - } - - let entries = list_recording_zip_entries(&recording_dir).await.map_err( - HttpError::internal() - .with_msg("failed to read recording manifest") - .err(), - )?; - - if entries.is_empty() { + if !recording_dir.is_dir() { return Err(HttpError::not_found().msg("requested recording does not exist")); } - // Bounded buffer: backpressure when the client is slower than zip production. - let (zip_writer, zip_reader) = tokio::io::duplex(64 * 1024); - - tokio::spawn(async move { - if let Err(error) = write_recording_zip(zip_writer, &recording_dir, &entries).await { - warn!( - error = format!("{error:#}"), - session.id = %id, - "Failed to stream recording ZIP archive" - ); + let entries = match list_recording_zip_entries(&recording_dir).await { + Ok(entries) => entries, + Err(ListRecordingZipError::NotFound) => { + return Err(HttpError::not_found().msg("requested recording does not exist")); } - }); + Err(ListRecordingZipError::Other(error)) => { + return Err(HttpError::internal() + .with_msg("failed to read recording manifest") + .build(error)); + } + }; - let mut response = AsyncReadBody::new(zip_reader).into_response(); + let body = recording_zip_body(recording_dir, entries, id, shutdown_signal); + let mut response = Response::new(body); response .headers_mut() @@ -573,6 +576,10 @@ where { use tower::ServiceExt as _; + if claims.jet_rop != RecordingOperation::Pull { + return Err(HttpError::forbidden().msg("expected pull operation")); + } + if !is_safe_recording_file_name(&filename) { return Err(HttpError::bad_request().msg("invalid file name")); } @@ -587,7 +594,7 @@ where .join(id.to_string()) .join(filename); - if !path.exists() || !path.is_file() { + if !path.is_file() { return Err(HttpError::not_found().msg("requested file does not exist")); } @@ -623,24 +630,46 @@ struct RecordingZipManifestFile { file_name: String, } +#[derive(Debug)] +enum ListRecordingZipError { + /// Session directory has no usable `recording.json`. + NotFound, + /// Unexpected I/O failure while reading the manifest. + Other(anyhow::Error), +} + fn is_safe_recording_file_name(file_name: &str) -> bool { !file_name.is_empty() && !file_name.contains("..") && !file_name.contains('/') && !file_name.contains('\\') } /// Builds the ordered list of files to put in a session ZIP. /// -/// Always starts with `recording.json`, then appends every manifest entry that -/// exists on disk and uses a safe relative file name. -async fn list_recording_zip_entries(recording_dir: &Utf8Path) -> anyhow::Result> { +/// Always starts with `recording.json`, then appends every manifest entry that exists on disk and uses a safe relative file name. +async fn list_recording_zip_entries(recording_dir: &Utf8Path) -> Result, ListRecordingZipError> { let manifest_path = recording_dir.join("recording.json"); if !manifest_path.is_file() { - anyhow::bail!("recording manifest not found"); + return Err(ListRecordingZipError::NotFound); } - let manifest_json = tokio::fs::read(&manifest_path) - .await - .with_context(|| format!("read recording manifest at {manifest_path}"))?; - let manifest: RecordingZipManifest = serde_json::from_slice(&manifest_json).context("parse recording manifest")?; + let manifest_json = tokio::fs::read(&manifest_path).await.map_err(|error| { + if error.kind() == io::ErrorKind::NotFound { + ListRecordingZipError::NotFound + } else { + ListRecordingZipError::Other( + anyhow::Error::new(error).context(format!("read recording manifest at {manifest_path}")), + ) + } + })?; + + let manifest: RecordingZipManifest = serde_json::from_slice(&manifest_json).map_err(|error| { + // Corrupt/incomplete package: treat as missing recording for the pull contract. + debug!( + error = format!("{error:#}"), + path = %manifest_path, + "Invalid recording manifest" + ); + ListRecordingZipError::NotFound + })?; let mut entries = Vec::with_capacity(manifest.files.len() + 1); entries.push("recording.json".to_owned()); @@ -669,20 +698,102 @@ async fn list_recording_zip_entries(recording_dir: &Utf8Path) -> anyhow::Result< Ok(entries) } +/// Streams a ZIP body that fails the HTTP transfer if packaging aborts mid-stream. +/// +/// A clean EOF is only produced after a successful archive finish. +/// Writer errors and shutdown yield a stream `Err` so the client does not treat a truncated ZIP as success. +fn recording_zip_body( + recording_dir: Utf8PathBuf, + entries: Vec, + session_id: Uuid, + mut shutdown_signal: ShutdownSignal, +) -> Body { + let (zip_writer, zip_reader) = tokio::io::duplex(ZIP_CHUNK_SIZE); + let (result_tx, result_rx) = oneshot::channel::>(); + + tokio::spawn(async move { + let result = tokio::select! { + result = write_recording_zip(zip_writer, &recording_dir, &entries) => { + result.map_err(|error| format!("{error:#}")) + } + _ = shutdown_signal.wait() => { + Err("gateway shutdown while streaming recording ZIP".to_owned()) + } + }; + + if let Err(error) = &result { + warn!( + error, + session.id = %session_id, + "Failed to stream recording ZIP archive" + ); + } + + let _ = result_tx.send(result); + }); + + Body::from_stream(zip_body_stream(zip_reader, result_rx)) +} + +fn zip_body_stream( + reader: DuplexStream, + result_rx: oneshot::Receiver>, +) -> impl stream::Stream> { + stream::unfold( + ZipBodyState { + reader, + result_rx: Some(result_rx), + buffer: vec![0u8; ZIP_CHUNK_SIZE], + finished: false, + }, + |mut state| async move { + if state.finished { + return None; + } + + match state.reader.read(&mut state.buffer).await { + Ok(0) => { + let outcome = match state.result_rx.take() { + Some(result_rx) => result_rx + .await + .unwrap_or_else(|_| Err("recording ZIP task ended unexpectedly".to_owned())), + None => Ok(()), + }; + state.finished = true; + match outcome { + Ok(()) => None, + Err(message) => Some((Err(io::Error::other(message)), state)), + } + } + Ok(n) => { + let chunk = Bytes::copy_from_slice(&state.buffer[..n]); + Some((Ok(chunk), state)) + } + Err(error) => { + state.finished = true; + Some((Err(error), state)) + } + } + }, + ) +} + +struct ZipBodyState { + reader: DuplexStream, + result_rx: Option>>, + buffer: Vec, + finished: bool, +} + /// Streams a ZIP archive for the given recording files into `writer`. /// -/// Entries use the STORED method: recording payloads are already compressed (WebM, etc.), -/// so deflate would mainly burn CPU for little size gain. -async fn write_recording_zip(writer: W, recording_dir: &Utf8Path, entries: &[String]) -> anyhow::Result<()> -where - W: AsyncWrite + Unpin, -{ +/// Entries use the STORED method: recording payloads are already compressed (WebM, etc.), so deflate would mainly burn CPU for little size gain. +async fn write_recording_zip(writer: DuplexStream, recording_dir: &Utf8Path, entries: &[String]) -> anyhow::Result<()> { use futures::AsyncWriteExt as _; - use tokio::io::AsyncReadExt as _; let mut zip_writer = ZipFileWriter::with_tokio(writer); // Reused across entries to avoid per-file allocation for large multi-clip downloads. - let mut buffer = vec![0u8; 64 * 1024]; + let mut buffer = vec![0u8; ZIP_CHUNK_SIZE]; for file_name in entries { let path = recording_dir.join(file_name); @@ -724,15 +835,27 @@ where /// Collects a streamed ZIP into memory (tests / small fixtures only). #[cfg(test)] async fn collect_recording_zip(recording_dir: &Utf8Path, entries: &[String]) -> anyhow::Result> { - let (writer, mut reader) = tokio::io::duplex(64 * 1024); + use futures::StreamExt as _; + + let (writer, reader) = tokio::io::duplex(ZIP_CHUNK_SIZE); + let (result_tx, result_rx) = oneshot::channel(); let dir = recording_dir.to_owned(); let entries = entries.to_owned(); - let writer_task = tokio::spawn(async move { write_recording_zip(writer, &dir, &entries).await }); + let writer_task = tokio::spawn(async move { + let result = write_recording_zip(writer, &dir, &entries) + .await + .map_err(|error| format!("{error:#}")); + let _ = result_tx.send(result); + }); let mut bytes = Vec::new(); - tokio::io::copy(&mut reader, &mut bytes).await?; - writer_task.await.context("zip writer task join")??; + let mut stream = std::pin::pin!(zip_body_stream(reader, result_rx)); + while let Some(chunk) = stream.next().await { + bytes.extend_from_slice(&chunk.context("read ZIP body chunk")?); + } + + writer_task.await.context("zip writer task join")?; Ok(bytes) } @@ -937,4 +1060,65 @@ mod tests { assert_eq!(content, expected); } } + + #[tokio::test] + async fn missing_manifest_is_not_found() { + let dir = tempfile::tempdir().expect("temp dir"); + let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); + + let error = list_recording_zip_entries(&dir_path) + .await + .expect_err("missing manifest"); + assert!(matches!(error, ListRecordingZipError::NotFound)); + } + + #[tokio::test] + async fn corrupt_manifest_is_not_found() { + let dir = tempfile::tempdir().expect("temp dir"); + let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); + + tokio::fs::write(dir_path.join("recording.json"), b"{not-json") + .await + .expect("write corrupt manifest"); + + let error = list_recording_zip_entries(&dir_path) + .await + .expect_err("corrupt manifest"); + assert!(matches!(error, ListRecordingZipError::NotFound)); + } + + #[tokio::test] + async fn mid_stream_open_failure_errors_the_body() { + let dir = tempfile::tempdir().expect("temp dir"); + let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); + + let manifest = serde_json::json!({ + "sessionId": "33333333-3333-3333-3333-333333333333", + "startTime": 1, + "duration": 2, + "files": [ + { "fileName": "recording-0.webm", "startTime": 1, "duration": 2 } + ] + }); + + tokio::fs::write(dir_path.join("recording.json"), manifest.to_string()) + .await + .expect("write manifest"); + tokio::fs::write(dir_path.join("recording-0.webm"), b"clip") + .await + .expect("write clip"); + + // Pass a listed path that does not exist so packaging fails after headers would be sent. + let entries = vec!["recording.json".to_owned(), "missing-clip.webm".to_owned()]; + let error = collect_recording_zip(&dir_path, &entries) + .await + .expect_err("zip body should fail"); + assert!( + error.to_string().contains("read ZIP body chunk") + || error + .downcast_ref::() + .is_some_and(|io_error| io_error.kind() == io::ErrorKind::Other), + "unexpected error: {error:#}" + ); + } } From 454c11a3829165e7cc090fc6b343c456aba61a69 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 10 Aug 2026 15:20:55 -0400 Subject: [PATCH 3/4] fix(dgw): cap session recording ZIP downloads Refuse session package pulls that exceed 128 files or 2 GiB uncompressed with HTTP 413 so pathological manifests cannot pin the gateway on huge bulk transfers while normal multi-clip sessions stay unaffected. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- devolutions-gateway/openapi/gateway-api.yaml | 2 + devolutions-gateway/src/api/jrec.rs | 117 +++++++++++++++++++ 2 files changed, 119 insertions(+) diff --git a/devolutions-gateway/openapi/gateway-api.yaml b/devolutions-gateway/openapi/gateway-api.yaml index 4159240cf..0eacfe042 100644 --- a/devolutions-gateway/openapi/gateway-api.yaml +++ b/devolutions-gateway/openapi/gateway-api.yaml @@ -286,6 +286,8 @@ paths: description: Insufficient permissions '404': description: Recording not found + '413': + description: Recording package exceeds download size or file-count limits security: - jrec_token: - pull diff --git a/devolutions-gateway/src/api/jrec.rs b/devolutions-gateway/src/api/jrec.rs index 2c8424705..6717e3fa8 100644 --- a/devolutions-gateway/src/api/jrec.rs +++ b/devolutions-gateway/src/api/jrec.rs @@ -34,6 +34,17 @@ use crate::token::{JrecTokenClaims, RecordingFileType, RecordingOperation}; /// Read/write chunk size for session ZIP streaming (duplex buffer and file copy). const ZIP_CHUNK_SIZE: usize = 64 * 1024; +/// Maximum files in a session ZIP (`recording.json` + clips). +/// +/// Reconnect windows only mint a small number of clips per session in practice; +/// this bound blocks pathological manifests without rejecting normal multi-clip packages. +const MAX_RECORDING_ZIP_FILES: usize = 128; + +/// Maximum total uncompressed payload (bytes) for a session ZIP download. +/// +/// Chosen to cover multi-hour WebM packages with headroom while limiting concurrent bulk pulls. +const MAX_RECORDING_ZIP_BYTES: u64 = 2 * 1024 * 1024 * 1024; + pub fn make_router(state: DgwState) -> Router { Router::new() .route("/push/{id}", get(jrec_push)) @@ -492,6 +503,7 @@ pub(crate) async fn list_recordings( (status = 401, description = "Invalid or missing authorization token"), (status = 403, description = "Insufficient permissions"), (status = 404, description = "Recording not found"), + (status = 413, description = "Recording package exceeds download size or file-count limits"), ), security(("jrec_token" = ["pull"])), ))] @@ -530,6 +542,8 @@ pub(crate) async fn pull_recording_session( } }; + enforce_recording_zip_limits(&recording_dir, &entries).await?; + let body = recording_zip_body(recording_dir, entries, id, shutdown_signal); let mut response = Response::new(body); @@ -698,6 +712,69 @@ async fn list_recording_zip_entries(recording_dir: &Utf8Path) -> Result Option { + if file_count > MAX_RECORDING_ZIP_FILES { + Some(RecordingZipLimitKind::FileCount) + } else if total_bytes > MAX_RECORDING_ZIP_BYTES { + Some(RecordingZipLimitKind::TotalBytes) + } else { + None + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RecordingZipLimitKind { + FileCount, + TotalBytes, +} + +impl RecordingZipLimitKind { + const fn message(self) -> &'static str { + match self { + Self::FileCount => "recording package exceeds maximum file count for download", + Self::TotalBytes => "recording package exceeds maximum size for download", + } + } +} + +/// Rejects session packages that are too large to stream safely through this endpoint. +/// +/// Limits are evaluated up front from directory metadata so the client gets a clear HTTP error +/// instead of a multi-gigabyte transfer that may time out or pressure the host. +async fn enforce_recording_zip_limits(recording_dir: &Utf8Path, entries: &[String]) -> Result<(), HttpError> { + let mut total_bytes = 0u64; + for file_name in entries { + let path = recording_dir.join(file_name); + let metadata = tokio::fs::metadata(&path).await.map_err(|error| { + if error.kind() == io::ErrorKind::NotFound { + HttpError::not_found().msg("requested recording does not exist") + } else { + HttpError::internal() + .with_msg("failed to stat recording file for download limits") + .build(error) + } + })?; + total_bytes = total_bytes.saturating_add(metadata.len()); + } + + match recording_zip_limits_exceeded(entries.len(), total_bytes) { + None => Ok(()), + Some(kind) => { + warn!( + ?kind, + file_count = entries.len(), + total_bytes, + file_limit = MAX_RECORDING_ZIP_FILES, + byte_limit = MAX_RECORDING_ZIP_BYTES, + path = %recording_dir, + "Refusing recording ZIP download: package exceeds safety limits" + ); + Err(HttpErrorBuilder::new(StatusCode::PAYLOAD_TOO_LARGE).msg(kind.message())) + } + } +} + /// Streams a ZIP body that fails the HTTP transfer if packaging aborts mid-stream. /// /// A clean EOF is only produced after a successful archive finish. @@ -1121,4 +1198,44 @@ mod tests { "unexpected error: {error:#}" ); } + + #[test] + fn zip_limits_allow_typical_packages() { + assert_eq!(recording_zip_limits_exceeded(1, 0), None); + assert_eq!(recording_zip_limits_exceeded(2, 200 * 1024 * 1024), None); + assert_eq!( + recording_zip_limits_exceeded(MAX_RECORDING_ZIP_FILES, MAX_RECORDING_ZIP_BYTES), + None + ); + } + + #[test] + fn zip_limits_reject_pathological_packages() { + assert_eq!( + recording_zip_limits_exceeded(MAX_RECORDING_ZIP_FILES + 1, 1), + Some(RecordingZipLimitKind::FileCount) + ); + assert_eq!( + recording_zip_limits_exceeded(1, MAX_RECORDING_ZIP_BYTES + 1), + Some(RecordingZipLimitKind::TotalBytes) + ); + } + + #[tokio::test] + async fn enforce_limits_rejects_too_many_files() { + let dir = tempfile::tempdir().expect("temp dir"); + let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); + + let mut entries = Vec::with_capacity(MAX_RECORDING_ZIP_FILES + 1); + for index in 0..=MAX_RECORDING_ZIP_FILES { + let name = format!("f-{index}.bin"); + tokio::fs::write(dir_path.join(&name), b"x").await.expect("write file"); + entries.push(name); + } + + let error = enforce_recording_zip_limits(&dir_path, &entries) + .await + .expect_err("too many files"); + assert_eq!(error.code, StatusCode::PAYLOAD_TOO_LARGE); + } } From 2013b996df53acde63952db30c7af6de64b3e9de Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 10 Aug 2026 16:36:13 -0400 Subject: [PATCH 4/4] fix(dgw): snapshot recording ZIPs with interoperable headers Address review feedback on session package downloads: freeze the recording.json bytes that define membership so reconnects cannot widen the archive mid-download, and build STORED entries with the zip crate so local-file headers carry real sizes/CRCs accepted by common unzip tools. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- Cargo.lock | 70 ++-- devolutions-gateway/Cargo.toml | 3 +- devolutions-gateway/src/api/jrec.rs | 533 +++++++++++++++------------- 3 files changed, 324 insertions(+), 282 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 40994c9bd..e437ddb96 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -201,6 +201,15 @@ version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" +[[package]] +name = "arbitrary" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1" +dependencies = [ + "derive_arbitrary", +] + [[package]] name = "arc-swap" version = "1.9.2" @@ -383,20 +392,6 @@ dependencies = [ "syn 2.0.118", ] -[[package]] -name = "async_zip" -version = "0.0.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d8c50d65ce1b0e0cb65a785ff615f78860d7754290647d3b983208daa4f85e6" -dependencies = [ - "crc32fast", - "futures-lite", - "pin-project 1.1.13", - "thiserror 2.0.18", - "tokio 1.52.3", - "tokio-util", -] - [[package]] name = "atomic-polyfill" version = "1.0.3" @@ -1642,6 +1637,17 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "derive_arbitrary" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e567bd82dcff979e4b03460c307b3cdc9e96fde3d73bed1496d2bc75d9dd62a" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.46", + "syn 2.0.118", +] + [[package]] name = "derive_more" version = "2.1.1" @@ -1778,7 +1784,6 @@ dependencies = [ "anyhow", "argon2", "async-trait", - "async_zip", "axum 0.8.9", "axum-extra", "backoff", @@ -1873,6 +1878,7 @@ dependencies = [ "windows-sys 0.61.2", "x509-cert 0.3.0", "zeroize", + "zip", ] [[package]] @@ -2551,19 +2557,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" -[[package]] -name = "futures-lite" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" -dependencies = [ - "fastrand", - "futures-core", - "futures-io", - "parking", - "pin-project-lite 0.2.17", -] - [[package]] name = "futures-macro" version = "0.3.32" @@ -5084,12 +5077,6 @@ dependencies = [ "sha2 0.11.0", ] -[[package]] -name = "parking" -version = "2.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" - [[package]] name = "parking_lot" version = "0.12.5" @@ -9706,6 +9693,21 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "zip" +version = "2.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fabe6324e908f85a1c52063ce7aa26b68dcb7eb6dbc83a2d148403c9bc3eba50" +dependencies = [ + "arbitrary", + "crc32fast", + "crossbeam-utils", + "displaydoc", + "indexmap 2.14.0", + "memchr", + "thiserror 2.0.18", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/devolutions-gateway/Cargo.toml b/devolutions-gateway/Cargo.toml index 9143dd41f..e0bafdce1 100644 --- a/devolutions-gateway/Cargo.toml +++ b/devolutions-gateway/Cargo.toml @@ -107,7 +107,8 @@ tungstenite = "0.29" # Should be the same version as `axum` (we perform error do tokio-tungstenite = { version = "0.29", features = ["rustls-tls-native-roots"] } # Should use the same version of tungstenite as `axum` http-body-util = "0.1" tokio-retry = "0.3" -async_zip = { version = "0.0.18", default-features = false, features = ["tokio"] } +zip = { version = "2.4", default-features = false } +tempfile = "3" # OpenAPI generator utoipa = { version = "4.2", default-features = false, features = ["uuid", "time"], optional = true } diff --git a/devolutions-gateway/src/api/jrec.rs b/devolutions-gateway/src/api/jrec.rs index 6717e3fa8..b010f13a6 100644 --- a/devolutions-gateway/src/api/jrec.rs +++ b/devolutions-gateway/src/api/jrec.rs @@ -1,11 +1,12 @@ +use std::fs; +use std::io::{self, Seek as _, Write as _}; use std::net::SocketAddr; -use std::path::Path; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; -use std::{fs, io}; use anyhow::Context as _; -use async_zip::tokio::write::ZipFileWriter; -use async_zip::{Compression, ZipEntryBuilder}; use axum::body::Body; use axum::extract::ws::{CloseFrame, WebSocket}; use axum::extract::{self, ConnectInfo, Query, State, WebSocketUpgrade}; @@ -19,10 +20,11 @@ use camino::{Utf8Path, Utf8PathBuf}; use devolutions_gateway_task::ShutdownSignal; use futures::stream; use hyper::StatusCode; -use tokio::io::{AsyncReadExt as _, DuplexStream}; -use tokio::sync::oneshot; +use tokio::io::AsyncReadExt as _; use tracing::Instrument as _; use uuid::Uuid; +use zip::CompressionMethod; +use zip::write::SimpleFileOptions; use crate::DgwState; use crate::api::heartbeat::recording_storage_health; @@ -31,7 +33,7 @@ use crate::http::{HttpError, HttpErrorBuilder}; use crate::recording::{PushOutcome, RecordingMessageSender}; use crate::token::{JrecTokenClaims, RecordingFileType, RecordingOperation}; -/// Read/write chunk size for session ZIP streaming (duplex buffer and file copy). +/// Read chunk size when streaming a finished session ZIP from the temp file. const ZIP_CHUNK_SIZE: usize = 64 * 1024; /// Maximum files in a session ZIP (`recording.json` + clips). @@ -530,21 +532,11 @@ pub(crate) async fn pull_recording_session( return Err(HttpError::not_found().msg("requested recording does not exist")); } - let entries = match list_recording_zip_entries(&recording_dir).await { - Ok(entries) => entries, - Err(ListRecordingZipError::NotFound) => { - return Err(HttpError::not_found().msg("requested recording does not exist")); - } - Err(ListRecordingZipError::Other(error)) => { - return Err(HttpError::internal() - .with_msg("failed to read recording manifest") - .build(error)); - } - }; - - enforce_recording_zip_limits(&recording_dir, &entries).await?; + // Snapshot membership once so a reconnect cannot widen the package after we start packaging. + let plan = snapshot_recording_zip_plan(&recording_dir).await?; + enforce_recording_zip_limits(&recording_dir, &plan).await?; - let body = recording_zip_body(recording_dir, entries, id, shutdown_signal); + let body = recording_zip_body(recording_dir, plan, id, shutdown_signal).await?; let mut response = Response::new(body); response @@ -644,50 +636,50 @@ struct RecordingZipManifestFile { file_name: String, } -#[derive(Debug)] -enum ListRecordingZipError { - /// Session directory has no usable `recording.json`. - NotFound, - /// Unexpected I/O failure while reading the manifest. - Other(anyhow::Error), -} - fn is_safe_recording_file_name(file_name: &str) -> bool { !file_name.is_empty() && !file_name.contains("..") && !file_name.contains('/') && !file_name.contains('\\') } -/// Builds the ordered list of files to put in a session ZIP. +/// Immutable package membership for one download attempt. /// -/// Always starts with `recording.json`, then appends every manifest entry that exists on disk and uses a safe relative file name. -async fn list_recording_zip_entries(recording_dir: &Utf8Path) -> Result, ListRecordingZipError> { - let manifest_path = recording_dir.join("recording.json"); - if !manifest_path.is_file() { - return Err(ListRecordingZipError::NotFound); +/// `manifest_bytes` are the exact `recording.json` contents used to derive `clip_names`, +/// so the archived manifest cannot drift from the clips included in the ZIP. +#[derive(Debug, Clone)] +struct RecordingZipPlan { + manifest_bytes: Vec, + clip_names: Vec, +} + +impl RecordingZipPlan { + fn entry_count(&self) -> usize { + 1 /* recording.json */ + self.clip_names.len() } +} - let manifest_json = tokio::fs::read(&manifest_path).await.map_err(|error| { +/// Snapshots `recording.json` and the clip files it references at call time. +async fn snapshot_recording_zip_plan(recording_dir: &Utf8Path) -> Result { + let manifest_path = recording_dir.join("recording.json"); + let manifest_bytes = tokio::fs::read(&manifest_path).await.map_err(|error| { if error.kind() == io::ErrorKind::NotFound { - ListRecordingZipError::NotFound + HttpError::not_found().msg("requested recording does not exist") } else { - ListRecordingZipError::Other( - anyhow::Error::new(error).context(format!("read recording manifest at {manifest_path}")), - ) + HttpError::internal() + .with_msg("failed to read recording manifest") + .build(anyhow::Error::new(error).context(format!("read recording manifest at {manifest_path}"))) } })?; - let manifest: RecordingZipManifest = serde_json::from_slice(&manifest_json).map_err(|error| { + let manifest: RecordingZipManifest = serde_json::from_slice(&manifest_bytes).map_err(|error| { // Corrupt/incomplete package: treat as missing recording for the pull contract. debug!( error = format!("{error:#}"), path = %manifest_path, "Invalid recording manifest" ); - ListRecordingZipError::NotFound + HttpError::not_found().msg("requested recording does not exist") })?; - let mut entries = Vec::with_capacity(manifest.files.len() + 1); - entries.push("recording.json".to_owned()); - + let mut clip_names = Vec::with_capacity(manifest.files.len()); for file in manifest.files { if !is_safe_recording_file_name(&file.file_name) { warn!( @@ -699,7 +691,7 @@ async fn list_recording_zip_entries(recording_dir: &Utf8Path) -> Result Result Option { if file_count > MAX_RECORDING_ZIP_FILES { Some(RecordingZipLimitKind::FileCount) @@ -742,9 +737,9 @@ impl RecordingZipLimitKind { /// /// Limits are evaluated up front from directory metadata so the client gets a clear HTTP error /// instead of a multi-gigabyte transfer that may time out or pressure the host. -async fn enforce_recording_zip_limits(recording_dir: &Utf8Path, entries: &[String]) -> Result<(), HttpError> { - let mut total_bytes = 0u64; - for file_name in entries { +async fn enforce_recording_zip_limits(recording_dir: &Utf8Path, plan: &RecordingZipPlan) -> Result<(), HttpError> { + let mut total_bytes = u64::try_from(plan.manifest_bytes.len()).unwrap_or(u64::MAX); + for file_name in &plan.clip_names { let path = recording_dir.join(file_name); let metadata = tokio::fs::metadata(&path).await.map_err(|error| { if error.kind() == io::ErrorKind::NotFound { @@ -758,12 +753,12 @@ async fn enforce_recording_zip_limits(recording_dir: &Utf8Path, entries: &[Strin total_bytes = total_bytes.saturating_add(metadata.len()); } - match recording_zip_limits_exceeded(entries.len(), total_bytes) { + match recording_zip_limits_exceeded(plan.entry_count(), total_bytes) { None => Ok(()), Some(kind) => { warn!( ?kind, - file_count = entries.len(), + file_count = plan.entry_count(), total_bytes, file_limit = MAX_RECORDING_ZIP_FILES, byte_limit = MAX_RECORDING_ZIP_BYTES, @@ -775,51 +770,118 @@ async fn enforce_recording_zip_limits(recording_dir: &Utf8Path, entries: &[Strin } } -/// Streams a ZIP body that fails the HTTP transfer if packaging aborts mid-stream. +/// Builds a complete, interoperable ZIP then streams it. /// -/// A clean EOF is only produced after a successful archive finish. -/// Writer errors and shutdown yield a stream `Err` so the client does not treat a truncated ZIP as success. -fn recording_zip_body( +/// Packaging uses the standard `zip` crate with known entry sizes/CRC so common OS unzippers accept the archive. +/// The snapshotted manifest bytes are written as-is (not re-read), keeping membership consistent. +async fn recording_zip_body( recording_dir: Utf8PathBuf, - entries: Vec, + plan: RecordingZipPlan, session_id: Uuid, mut shutdown_signal: ShutdownSignal, -) -> Body { - let (zip_writer, zip_reader) = tokio::io::duplex(ZIP_CHUNK_SIZE); - let (result_tx, result_rx) = oneshot::channel::>(); - +) -> Result { + let cancel = Arc::new(AtomicBool::new(false)); + let cancel_for_shutdown = Arc::clone(&cancel); tokio::spawn(async move { - let result = tokio::select! { - result = write_recording_zip(zip_writer, &recording_dir, &entries) => { - result.map_err(|error| format!("{error:#}")) - } - _ = shutdown_signal.wait() => { - Err("gateway shutdown while streaming recording ZIP".to_owned()) - } - }; + shutdown_signal.wait().await; + cancel_for_shutdown.store(true, Ordering::Relaxed); + }); - if let Err(error) = &result { + let recording_dir_for_build = PathBuf::from(recording_dir.as_std_path()); + let plan_for_build = plan; + let built = tokio::task::spawn_blocking(move || { + build_recording_zip_archive(&recording_dir_for_build, &plan_for_build, &cancel) + }) + .await + .map_err(|error| { + HttpError::internal() + .with_msg("recording ZIP worker failed") + .build(error) + })? + .map_err(|error| { + if error.root_cause().downcast_ref::().is_some() { + HttpErrorBuilder::new(StatusCode::SERVICE_UNAVAILABLE).msg("recording download cancelled") + } else { warn!( - error, + error = format!("{error:#}"), session.id = %session_id, - "Failed to stream recording ZIP archive" + "Failed to build recording ZIP archive" ); + HttpError::internal() + .with_msg("failed to build recording ZIP archive") + .build(error) } + })?; - let _ = result_tx.send(result); - }); + let (std_file, temp_path) = built.into_parts(); + let file = tokio::fs::File::from_std(std_file); + Ok(Body::from_stream(zip_file_body_stream(file, temp_path))) +} + +#[derive(Debug)] +struct RecordingZipCancelled; + +impl std::fmt::Display for RecordingZipCancelled { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("recording ZIP build cancelled") + } +} + +impl std::error::Error for RecordingZipCancelled {} + +/// Writes a STORED ZIP with valid local-file headers to a temporary file. +fn build_recording_zip_archive( + recording_dir: &Path, + plan: &RecordingZipPlan, + cancel: &AtomicBool, +) -> anyhow::Result { + if cancel.load(Ordering::Relaxed) { + return Err(anyhow::Error::new(RecordingZipCancelled)); + } + + let mut tmp = tempfile::NamedTempFile::new().context("create temp file for recording ZIP")?; + { + let mut zip = zip::ZipWriter::new(tmp.as_file_mut()); + let options = SimpleFileOptions::default().compression_method(CompressionMethod::Stored); - Body::from_stream(zip_body_stream(zip_reader, result_rx)) + zip.start_file("recording.json", options) + .context("start recording.json ZIP entry")?; + zip.write_all(&plan.manifest_bytes) + .context("write recording.json ZIP entry")?; + + for file_name in &plan.clip_names { + if cancel.load(Ordering::Relaxed) { + return Err(anyhow::Error::new(RecordingZipCancelled)); + } + + let path = recording_dir.join(file_name); + let mut file = + fs::File::open(&path).with_context(|| format!("open recording file at {}", path.display()))?; + + zip.start_file(file_name.as_str(), options) + .with_context(|| format!("start ZIP entry for {file_name}"))?; + io::copy(&mut file, &mut zip).with_context(|| format!("write ZIP entry for {file_name}"))?; + } + + zip.finish().context("finish ZIP archive")?; + } + + tmp.as_file_mut().sync_all().context("sync recording ZIP temp file")?; + tmp.as_file_mut() + .seek(io::SeekFrom::Start(0)) + .context("rewind recording ZIP temp file")?; + Ok(tmp) } -fn zip_body_stream( - reader: DuplexStream, - result_rx: oneshot::Receiver>, +fn zip_file_body_stream( + file: tokio::fs::File, + temp_path: tempfile::TempPath, ) -> impl stream::Stream> { stream::unfold( - ZipBodyState { - reader, - result_rx: Some(result_rx), + ZipFileBodyState { + file, + // Keep the temp path alive until the response body is fully consumed or dropped. + _temp_path: temp_path, buffer: vec![0u8; ZIP_CHUNK_SIZE], finished: false, }, @@ -828,19 +890,10 @@ fn zip_body_stream( return None; } - match state.reader.read(&mut state.buffer).await { + match state.file.read(&mut state.buffer).await { Ok(0) => { - let outcome = match state.result_rx.take() { - Some(result_rx) => result_rx - .await - .unwrap_or_else(|_| Err("recording ZIP task ended unexpectedly".to_owned())), - None => Ok(()), - }; state.finished = true; - match outcome { - Ok(()) => None, - Err(message) => Some((Err(io::Error::other(message)), state)), - } + None } Ok(n) => { let chunk = Bytes::copy_from_slice(&state.buffer[..n]); @@ -855,87 +908,13 @@ fn zip_body_stream( ) } -struct ZipBodyState { - reader: DuplexStream, - result_rx: Option>>, +struct ZipFileBodyState { + file: tokio::fs::File, + _temp_path: tempfile::TempPath, buffer: Vec, finished: bool, } -/// Streams a ZIP archive for the given recording files into `writer`. -/// -/// Entries use the STORED method: recording payloads are already compressed (WebM, etc.), so deflate would mainly burn CPU for little size gain. -async fn write_recording_zip(writer: DuplexStream, recording_dir: &Utf8Path, entries: &[String]) -> anyhow::Result<()> { - use futures::AsyncWriteExt as _; - - let mut zip_writer = ZipFileWriter::with_tokio(writer); - // Reused across entries to avoid per-file allocation for large multi-clip downloads. - let mut buffer = vec![0u8; ZIP_CHUNK_SIZE]; - - for file_name in entries { - let path = recording_dir.join(file_name); - let mut file = tokio::fs::File::open(&path) - .await - .with_context(|| format!("open recording file at {path}"))?; - - let builder = ZipEntryBuilder::new(file_name.clone().into(), Compression::Stored); - let mut entry_writer = zip_writer - .write_entry_stream(builder) - .await - .with_context(|| format!("start ZIP entry for {file_name}"))?; - - loop { - let n = file - .read(&mut buffer) - .await - .with_context(|| format!("read recording file at {path}"))?; - if n == 0 { - break; - } - - entry_writer - .write_all(&buffer[..n]) - .await - .with_context(|| format!("write ZIP entry for {file_name}"))?; - } - - entry_writer - .close() - .await - .with_context(|| format!("finish ZIP entry for {file_name}"))?; - } - - zip_writer.close().await.context("finish ZIP archive")?; - Ok(()) -} - -/// Collects a streamed ZIP into memory (tests / small fixtures only). -#[cfg(test)] -async fn collect_recording_zip(recording_dir: &Utf8Path, entries: &[String]) -> anyhow::Result> { - use futures::StreamExt as _; - - let (writer, reader) = tokio::io::duplex(ZIP_CHUNK_SIZE); - let (result_tx, result_rx) = oneshot::channel(); - let dir = recording_dir.to_owned(); - let entries = entries.to_owned(); - - let writer_task = tokio::spawn(async move { - let result = write_recording_zip(writer, &dir, &entries) - .await - .map_err(|error| format!("{error:#}")); - let _ = result_tx.send(result); - }); - - let mut bytes = Vec::new(); - let mut stream = std::pin::pin!(zip_body_stream(reader, result_rx)); - while let Some(chunk) = stream.next().await { - bytes.extend_from_slice(&chunk.context("read ZIP body chunk")?); - } - - writer_task.await.context("zip writer task join")?; - Ok(bytes) -} - async fn get_player( State(DgwState { conf_handle, .. }): State, path: Option>, @@ -1020,7 +999,10 @@ async fn shadow_recording( #[cfg(test)] mod tests { - use async_zip::base::read::mem::ZipFileReader; + use std::io::Read as _; + + use http_body_util::BodyExt as _; + use zip::ZipArchive; use super::*; @@ -1035,7 +1017,7 @@ mod tests { } #[tokio::test] - async fn lists_manifest_files_for_zip() { + async fn snapshots_manifest_files_for_zip() { let dir = tempfile::tempdir().expect("temp dir"); let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); @@ -1050,8 +1032,9 @@ mod tests { { "fileName": "../escape.webm", "startTime": 12, "duration": 1 } ] }); + let manifest_bytes = manifest.to_string().into_bytes(); - tokio::fs::write(dir_path.join("recording.json"), manifest.to_string()) + tokio::fs::write(dir_path.join("recording.json"), &manifest_bytes) .await .expect("write manifest"); tokio::fs::write(dir_path.join("recording-0.webm"), b"clip-zero") @@ -1061,19 +1044,78 @@ mod tests { .await .expect("write clip 1"); - let entries = list_recording_zip_entries(&dir_path).await.expect("list entries"); + let plan = snapshot_recording_zip_plan(&dir_path) + .await + .unwrap_or_else(|error| panic!("snapshot plan: {error}")); + assert_eq!(plan.manifest_bytes, manifest_bytes); assert_eq!( - entries, - vec![ - "recording.json".to_owned(), - "recording-0.webm".to_owned(), - "recording-1.webm".to_owned(), - ] + plan.clip_names, + vec!["recording-0.webm".to_owned(), "recording-1.webm".to_owned()] ); } #[tokio::test] - async fn streams_zip_with_all_listed_clips() { + async fn zip_keeps_snapshotted_manifest_when_disk_manifest_changes() { + let dir = tempfile::tempdir().expect("temp dir"); + let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); + + let original = serde_json::json!({ + "sessionId": "22222222-2222-2222-2222-222222222222", + "startTime": 1, + "duration": 2, + "files": [ + { "fileName": "recording-0.webm", "startTime": 1, "duration": 2 } + ] + }) + .to_string() + .into_bytes(); + + tokio::fs::write(dir_path.join("recording.json"), &original) + .await + .expect("write original manifest"); + tokio::fs::write(dir_path.join("recording-0.webm"), b"clip-zero") + .await + .expect("write clip 0"); + + let plan = snapshot_recording_zip_plan(&dir_path) + .await + .unwrap_or_else(|error| panic!("snapshot plan: {error}")); + + // Simulate a reconnect rewriting the live manifest after the download snapshot. + let updated = serde_json::json!({ + "sessionId": "22222222-2222-2222-2222-222222222222", + "startTime": 1, + "duration": 4, + "files": [ + { "fileName": "recording-0.webm", "startTime": 1, "duration": 2 }, + { "fileName": "recording-1.webm", "startTime": 3, "duration": 2 } + ] + }) + .to_string() + .into_bytes(); + tokio::fs::write(dir_path.join("recording.json"), &updated) + .await + .expect("rewrite manifest"); + tokio::fs::write(dir_path.join("recording-1.webm"), b"clip-one") + .await + .expect("write clip 1"); + + let archive = + build_recording_zip_archive(dir_path.as_std_path(), &plan, &AtomicBool::new(false)).expect("build zip"); + let mut zip = ZipArchive::new(archive.reopen().expect("reopen zip")).expect("open zip"); + assert_eq!(zip.len(), 2); + + let mut manifest_entry = zip.by_name("recording.json").expect("manifest entry"); + let mut archived_manifest = Vec::new(); + manifest_entry + .read_to_end(&mut archived_manifest) + .expect("read archived manifest"); + assert_eq!(archived_manifest, original); + assert_ne!(archived_manifest, updated); + } + + #[tokio::test] + async fn streams_interoperable_zip_with_all_listed_clips() { let dir = tempfile::tempdir().expect("temp dir"); let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); @@ -1098,43 +1140,34 @@ mod tests { .await .expect("write clip 1"); - let entries = list_recording_zip_entries(&dir_path).await.expect("list entries"); - let zip_bytes = collect_recording_zip(&dir_path, &entries).await.expect("build zip"); + let plan = snapshot_recording_zip_plan(&dir_path) + .await + .unwrap_or_else(|error| panic!("snapshot plan: {error}")); + let (shutdown_handle, shutdown_signal) = devolutions_gateway_task::ShutdownHandle::new(); + let body = recording_zip_body(dir_path.clone(), plan, Uuid::nil(), shutdown_signal) + .await + .unwrap_or_else(|error| panic!("build body: {error}")); + let zip_bytes = body.collect().await.expect("collect ZIP body").to_bytes().to_vec(); + drop(shutdown_handle); assert_eq!(&zip_bytes[..2], b"PK"); - let reader = ZipFileReader::new(zip_bytes).await.expect("parse zip"); - let names: Vec<_> = reader - .file() - .entries() - .iter() - .map(|entry| entry.filename().as_str().expect("utf8 name").to_owned()) - .collect(); - - assert_eq!( - names, - vec![ - "recording.json".to_owned(), - "recording-0.webm".to_owned(), - "recording-1.webm".to_owned(), - ] - ); - - for (index, expected) in [ - manifest_bytes.as_slice(), - b"first-clip".as_slice(), - b"second-clip".as_slice(), - ] - .into_iter() - .enumerate() - { - let mut entry_reader = reader.reader_with_entry(index).await.expect("entry reader"); + // Interop: standard zip crate reader (same class of local-header expectations as OS tools). + let cursor = io::Cursor::new(zip_bytes); + let mut archive = ZipArchive::new(cursor).expect("parse zip with standard reader"); + assert_eq!(archive.len(), 3); + + let expected = [ + ("recording.json", manifest_bytes.as_slice()), + ("recording-0.webm", b"first-clip".as_slice()), + ("recording-1.webm", b"second-clip".as_slice()), + ]; + for (name, payload) in expected { + let mut entry = archive.by_name(name).unwrap_or_else(|_| panic!("missing {name}")); let mut content = Vec::new(); - entry_reader - .read_to_end_checked(&mut content) - .await - .expect("read entry"); - assert_eq!(content, expected); + entry.read_to_end(&mut content).expect("read entry"); + assert_eq!(content, payload, "payload mismatch for {name}"); + assert_eq!(entry.compression(), CompressionMethod::Stored); } } @@ -1143,10 +1176,10 @@ mod tests { let dir = tempfile::tempdir().expect("temp dir"); let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); - let error = list_recording_zip_entries(&dir_path) + let error = snapshot_recording_zip_plan(&dir_path) .await .expect_err("missing manifest"); - assert!(matches!(error, ListRecordingZipError::NotFound)); + assert_eq!(error.code, StatusCode::NOT_FOUND); } #[tokio::test] @@ -1158,47 +1191,49 @@ mod tests { .await .expect("write corrupt manifest"); - let error = list_recording_zip_entries(&dir_path) + let error = snapshot_recording_zip_plan(&dir_path) .await .expect_err("corrupt manifest"); - assert!(matches!(error, ListRecordingZipError::NotFound)); + assert_eq!(error.code, StatusCode::NOT_FOUND); } #[tokio::test] - async fn mid_stream_open_failure_errors_the_body() { + async fn manifest_read_failure_is_internal_error_with_context() { let dir = tempfile::tempdir().expect("temp dir"); let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); - - let manifest = serde_json::json!({ - "sessionId": "33333333-3333-3333-3333-333333333333", - "startTime": 1, - "duration": 2, - "files": [ - { "fileName": "recording-0.webm", "startTime": 1, "duration": 2 } - ] - }); - - tokio::fs::write(dir_path.join("recording.json"), manifest.to_string()) - .await - .expect("write manifest"); - tokio::fs::write(dir_path.join("recording-0.webm"), b"clip") + tokio::fs::create_dir(dir_path.join("recording.json")) .await - .expect("write clip"); + .expect("create manifest directory"); - // Pass a listed path that does not exist so packaging fails after headers would be sent. - let entries = vec!["recording.json".to_owned(), "missing-clip.webm".to_owned()]; - let error = collect_recording_zip(&dir_path, &entries) + let error = snapshot_recording_zip_plan(&dir_path) .await - .expect_err("zip body should fail"); + .expect_err("manifest read should fail"); + assert_eq!(error.code, StatusCode::INTERNAL_SERVER_ERROR); assert!( - error.to_string().contains("read ZIP body chunk") - || error - .downcast_ref::() - .is_some_and(|io_error| io_error.kind() == io::ErrorKind::Other), - "unexpected error: {error:#}" + error + .source + .as_deref() + .is_some_and(|source| source.to_string().contains("read recording manifest at")), + "unexpected error: {error}" ); } + #[tokio::test] + async fn missing_clip_during_build_fails_before_body() { + let dir = tempfile::tempdir().expect("temp dir"); + let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); + + let plan = RecordingZipPlan { + manifest_bytes: b"{}".to_vec(), + clip_names: vec!["missing-clip.webm".to_owned()], + }; + let (_shutdown_handle, shutdown_signal) = devolutions_gateway_task::ShutdownHandle::new(); + let error = recording_zip_body(dir_path, plan, Uuid::nil(), shutdown_signal) + .await + .expect_err("missing clip should fail packaging"); + assert_eq!(error.code, StatusCode::INTERNAL_SERVER_ERROR); + } + #[test] fn zip_limits_allow_typical_packages() { assert_eq!(recording_zip_limits_exceeded(1, 0), None); @@ -1226,14 +1261,18 @@ mod tests { let dir = tempfile::tempdir().expect("temp dir"); let dir_path = Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).expect("utf8 path"); - let mut entries = Vec::with_capacity(MAX_RECORDING_ZIP_FILES + 1); - for index in 0..=MAX_RECORDING_ZIP_FILES { + let mut clip_names = Vec::with_capacity(MAX_RECORDING_ZIP_FILES); + for index in 0..MAX_RECORDING_ZIP_FILES { let name = format!("f-{index}.bin"); tokio::fs::write(dir_path.join(&name), b"x").await.expect("write file"); - entries.push(name); + clip_names.push(name); } - let error = enforce_recording_zip_limits(&dir_path, &entries) + let plan = RecordingZipPlan { + manifest_bytes: b"{}".to_vec(), + clip_names, + }; + let error = enforce_recording_zip_limits(&dir_path, &plan) .await .expect_err("too many files"); assert_eq!(error.code, StatusCode::PAYLOAD_TOO_LARGE);