1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
using System.Buffers.Binary;
using System.Text.Json;
using System.Text.Json.Serialization;
namespace Agent.Admin.Client;
public sealed record AdminNetworkEndpoint(string Address, int Port);
public sealed record AdminRequest(
int Version,
string Operation,
Guid TaskId,
string? ApplicationSid = null,
string? NetworkAccess = null,
IReadOnlyList<AdminNetworkEndpoint>? Endpoints = null,
bool? OriginallyLoopbackExempt = null);
public sealed record AdminResponse(bool Success, string? ErrorCode = null, string? Message = null, bool? LoopbackExempt = null);
public static class AdminProtocol
{
public const int Version = 1;
public const int MaxMessageBytes = 64 * 1024;
private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web)
{
UnmappedMemberHandling = JsonUnmappedMemberHandling.Disallow
};
public static async Task WriteAsync<T>(Stream stream, T value, CancellationToken cancellationToken)
{
byte[] payload = JsonSerializer.SerializeToUtf8Bytes(value, Json);
if (payload.Length is <= 0 or > MaxMessageBytes) throw new InvalidDataException("Administrative pipe message exceeds the fixed size limit.");
byte[] header = new byte[sizeof(int)];
BinaryPrimitives.WriteInt32LittleEndian(header, payload.Length);
await stream.WriteAsync(header, cancellationToken).ConfigureAwait(false);
await stream.WriteAsync(payload, cancellationToken).ConfigureAwait(false);
await stream.FlushAsync(cancellationToken).ConfigureAwait(false);
}
public static async Task<T> ReadAsync<T>(Stream stream, CancellationToken cancellationToken)
{
byte[] header = new byte[sizeof(int)];
await ReadExactlyAsync(stream, header, cancellationToken).ConfigureAwait(false);
int length = BinaryPrimitives.ReadInt32LittleEndian(header);
if (length is <= 0 or > MaxMessageBytes) throw new InvalidDataException("Administrative pipe message has an invalid size.");
byte[] payload = new byte[length];
await ReadExactlyAsync(stream, payload, cancellationToken).ConfigureAwait(false);
return JsonSerializer.Deserialize<T>(payload, Json) ?? throw new InvalidDataException("Administrative pipe message is empty.");
}
private static async Task ReadExactlyAsync(Stream stream, Memory<byte> buffer, CancellationToken cancellationToken)
{
int offset = 0;
while (offset < buffer.Length)
{
int read = await stream.ReadAsync(buffer[offset..], cancellationToken).ConfigureAwait(false);
if (read == 0) throw new EndOfStreamException("Administrative pipe disconnected.");
offset += read;
}
}
}
|
