1package brokerwire
2
3// Shared semantic converters.
4//
5// Conversion mirrors the port, domain, and catalogue types losslessly and
6// validates every narrowing numeric cast before it happens. Wire uint32
7// taxonomy fields never truncate into a smaller semantic enum: values
8// whose high bits would alias a legitimate code are refused with
9// errConvertRange before any cast. Identity lengths are exact (16-byte
10// connection/operation/incarnation/lifecycle values); text fields enforce
11// UTF-8, control/bidi, and byte bounds.
12
13import (
14 "math"
15
16 "github.com/bnema/vev/internal/adapters/protoconv"
17 "github.com/bnema/vev/internal/domain"
18 "github.com/bnema/vev/internal/ports"
19 "github.com/bnema/vev/internal/protocol"
20 "github.com/bnema/vev/internal/protocol/catalogue"
21 "github.com/bnema/vev/internal/protocol/wire"
22)
23
24// brokerEnum8 maps shared narrowing failures to the local range sentinel.
25func brokerEnum8[T ~uint8](value uint32) (T, error) {
26 v, err := protoconv.Uint8[T](value)
27 if err != nil {
28 return 0, errConvertRange
29 }
30 return v, nil
31}
32
33// brokerEnum16 maps shared narrowing failures to the local range sentinel.
34func brokerEnum16[T ~uint16](value uint32) (T, error) {
35 v, err := protoconv.Uint16[T](value)
36 if err != nil {
37 return 0, errConvertRange
38 }
39 return v, nil
40}
41
42func scopeToWire(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID) *wire.BrokerScope {
43 id := append([]byte(nil), connection[:]...)
44 return &wire.BrokerScope{BrokerEpoch: uint64(epoch), ConnectionId: id}
45}
46
47func scopeFromWire(message *wire.BrokerScope) (ports.BrokerEpoch, ports.BrokerConnectionID, error) {
48 var connection ports.BrokerConnectionID
49 if message == nil {
50 return 0, connection, errConvertRange
51 }
52 if message.GetBrokerEpoch() == 0 {
53 return 0, connection, errConvertRange
54 }
55 raw := message.GetConnectionId()
56 if len(raw) != len(connection) {
57 return 0, connection, errConvertRange
58 }
59 copy(connection[:], raw)
60 if err := connection.Validate(); err != nil {
61 return 0, ports.BrokerConnectionID{}, errConvertRange
62 }
63 return ports.BrokerEpoch(message.GetBrokerEpoch()), connection, nil
64}
65
66func refToWire(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID, stream ports.BrokerStreamID) *wire.BrokerStreamRef {
67 return &wire.BrokerStreamRef{Scope: scopeToWire(epoch, connection), StreamId: uint64(stream)}
68}
69
70func refFromWire(message *wire.BrokerStreamRef) (ports.BrokerEpoch, ports.BrokerConnectionID, ports.BrokerStreamID, error) {
71 var connection ports.BrokerConnectionID
72 if message == nil {
73 return 0, connection, 0, errConvertRange
74 }
75 epoch, conn, err := scopeFromWire(message.GetScope())
76 if err != nil {
77 return 0, connection, 0, err
78 }
79 stream := ports.BrokerStreamID(message.GetStreamId())
80 if err := stream.Validate(); err != nil {
81 return 0, connection, 0, errConvertRange
82 }
83 return epoch, conn, stream, nil
84}
85
86func operationToWire(id ports.BrokerOperationID) []byte {
87 return append([]byte(nil), id[:]...)
88}
89
90func operationFromWire(raw []byte) (ports.BrokerOperationID, error) {
91 var id ports.BrokerOperationID
92 if len(raw) != len(id) {
93 return id, errConvertRange
94 }
95 copy(id[:], raw)
96 if err := id.Validate(); err != nil {
97 return ports.BrokerOperationID{}, errConvertRange
98 }
99 return id, nil
100}
101
102func registrationToWire(v domain.RemoteRegistration) *wire.RemoteRegistration {
103 return protoconv.BrokerRegistrationToWire(v)
104}
105func registrationFromWire(v *wire.RemoteRegistration) (domain.RemoteRegistration, error) {
106 out, err := protoconv.BrokerRegistrationFromWire(v)
107 if err != nil {
108 return domain.RemoteRegistration{}, errConvertRange
109 }
110 return out, nil
111}
112
113func exactTargetToWire(v protocol.ExactSessionTarget) *wire.ExactTarget {
114 return protoconv.BrokerExactTargetToWire(v)
115}
116
117func exactTargetFromWire(v *wire.ExactTarget) (protocol.ExactSessionTarget, error) {
118 out, err := protoconv.BrokerExactTargetFromWire(v)
119 if err != nil {
120 return protocol.ExactSessionTarget{}, errConvertRange
121 }
122 return out, nil
123}
124
125func policyToWire(v ports.BrokerPolicy) *wire.BrokerWirePolicy {
126 return protoconv.BrokerPolicyToWire(v)
127}
128
129func policyFromWire(v *wire.BrokerWirePolicy) (ports.BrokerPolicy, error) {
130 out, err := protoconv.BrokerPolicyFromWire(v)
131 if err != nil {
132 return ports.BrokerPolicy{}, mapBrokerConversionError(err)
133 }
134 return out, nil
135}
136
137func mapBrokerConversionError(err error) error {
138 switch err {
139 case protoconv.ErrOutOfRange:
140 return errConvertRange
141 case protoconv.ErrTooLarge:
142 return ErrTooLarge
143 default:
144 return ErrInvalidMessage
145 }
146}
147
148func validateBrokerDisplayText(v string, max int) error {
149 return mapOptionalBrokerError(protoconv.BrokerDisplayText(v, max))
150}
151
152func mapOptionalBrokerError(err error) error {
153 if err == nil {
154 return nil
155 }
156 return mapBrokerConversionError(err)
157}
158
159// validateBrokerEndpoint enforces the SSH target contract shared by
160// AddHost/RemoveHost/OpenStream/snapshot hosts.
161func validateBrokerEndpoint(endpoint string) error {
162 if err := domain.ValidateRemoteHostTarget(endpoint); err != nil {
163 return ErrInvalidMessage
164 }
165 return nil
166}
167
168// validateBrokerOrigin enforces presentation-only display-origin text. It
169// may be empty (the endpoint renders instead) but never carries control
170// or bidi characters.
171func validateBrokerOrigin(origin string) error {
172 if origin == "" {
173 return nil
174 }
175 if err := domain.ValidateRemoteDisplayOrigin(origin); err != nil {
176 return ErrInvalidMessage
177 }
178 return nil
179}
180
181func errorDetailToWire(detail ErrorDetail) *wire.BrokerErrorDetail {
182 return &wire.BrokerErrorDetail{
183 Code: uint32(detail.Code),
184 Text: detail.Text,
185 AdmissionCode: detail.AdmissionCode,
186 FailureKind: uint32(detail.FailureKind),
187 }
188}
189
190func errorDetailFromWire(message *wire.BrokerErrorDetail) (ErrorDetail, error) {
191 var detail ErrorDetail
192 if message == nil {
193 return detail, errConvertRange
194 }
195 code, err := brokerEnum8[ports.BrokerErrorCode](message.GetCode())
196 if err != nil {
197 return ErrorDetail{}, err
198 }
199 if err := code.Validate(); err != nil {
200 return ErrorDetail{}, ErrInvalidMessage
201 }
202 detail.Code = code
203 if err := validateBrokerDisplayText(message.GetText(), ports.BrokerMaxErrorBytes); err != nil {
204 return ErrorDetail{}, err
205 }
206 detail.Text = message.GetText()
207 admission := message.GetAdmissionCode()
208 if admission > MaxBrokerAdmissionCode {
209 return ErrorDetail{}, errConvertRange
210 }
211 detail.AdmissionCode = admission
212 kind, err := brokerEnum8[domain.RemoteFailureKind](message.GetFailureKind())
213 if err != nil {
214 return ErrorDetail{}, err
215 }
216 if kind > MaxBrokerFailureKind {
217 return ErrorDetail{}, ErrInvalidMessage
218 }
219 detail.FailureKind = kind
220 failure := ports.BrokerError{Code: code, Text: detail.Text}
221 if err := failure.Validate(); err != nil {
222 return ErrorDetail{}, ErrInvalidMessage
223 }
224 return detail, nil
225}
226
227func catalogTabToWire(tab catalogue.RemoteCatalogTab) *wire.BrokerCatalogTab {
228 return &wire.BrokerCatalogTab{
229 Id: tab.ID,
230 Index: uint32(tab.Index),
231 Name: tab.Name,
232 Detail: tab.Detail,
233 Attention: tab.Attention,
234 }
235}
236
237func catalogTabFromWire(message *wire.BrokerCatalogTab) (catalogue.RemoteCatalogTab, error) {
238 var tab catalogue.RemoteCatalogTab
239 if message == nil {
240 return tab, errConvertRange
241 }
242 if len(message.GetId()) > catalogue.RemoteCatalogMaxIDBytes ||
243 len(message.GetName()) > catalogue.RemoteCatalogMaxNameBytes ||
244 len(message.GetDetail()) > catalogue.RemoteCatalogMaxDetailBytes {
245 return catalogue.RemoteCatalogTab{}, ErrTooLarge
246 }
247 index := message.GetIndex()
248 if index >= catalogue.RemoteCatalogMaxTabsPerSess {
249 return catalogue.RemoteCatalogTab{}, errConvertRange
250 }
251 tab = catalogue.RemoteCatalogTab{
252 ID: message.GetId(),
253 Index: uint16(index),
254 Name: message.GetName(),
255 Detail: message.GetDetail(),
256 Attention: message.GetAttention(),
257 }
258 return tab, nil
259}
260
261func catalogSessionToWire(session catalogue.RemoteCatalogSession) *wire.BrokerCatalogSession {
262 lifecycle := append([]byte(nil), session.LifecycleID[:]...)
263 var state uint32
264 switch session.State {
265 case catalogue.RemoteCatalogSessionUp:
266 state = 1
267 case catalogue.RemoteCatalogSessionDown:
268 state = 2
269 case catalogue.RemoteCatalogSessionBroken:
270 state = 3
271 }
272 tabs := make([]*wire.BrokerCatalogTab, 0, len(session.Tabs))
273 for _, tab := range session.Tabs {
274 tabs = append(tabs, catalogTabToWire(tab))
275 }
276 out := &wire.BrokerCatalogSession{
277 LifecycleId: lifecycle,
278 Name: session.Name,
279 State: state,
280 Ephemeral: session.Ephemeral,
281 Attached: session.Attached,
282 LastUsedSeq: session.LastUsedSeq,
283 ActiveTabId: session.ActiveTabID,
284 Reason: session.Reason,
285 }
286 if len(session.Tabs) > 0 || session.Tabs != nil {
287 out.Tabs = tabs
288 }
289 return out
290}
291
292func catalogSessionFromWire(message *wire.BrokerCatalogSession) (catalogue.RemoteCatalogSession, error) {
293 var session catalogue.RemoteCatalogSession
294 if message == nil {
295 return session, errConvertRange
296 }
297 raw := message.GetLifecycleId()
298 if len(raw) != len(session.LifecycleID) {
299 return catalogue.RemoteCatalogSession{}, errConvertRange
300 }
301 copy(session.LifecycleID[:], raw)
302 var state catalogue.RemoteCatalogSessionState
303 switch message.GetState() {
304 case 1:
305 state = catalogue.RemoteCatalogSessionUp
306 case 2:
307 state = catalogue.RemoteCatalogSessionDown
308 case 3:
309 state = catalogue.RemoteCatalogSessionBroken
310 default:
311 return catalogue.RemoteCatalogSession{}, errConvertRange
312 }
313 session.State = state
314 session.Name = message.GetName()
315 session.Ephemeral = message.GetEphemeral()
316 session.Attached = message.GetAttached()
317 session.LastUsedSeq = message.GetLastUsedSeq()
318 session.ActiveTabID = message.GetActiveTabId()
319 session.Reason = message.GetReason()
320 wireTabs := message.GetTabs()
321 // Proto3 repeated fields carry no presence: an absent or empty tab list
322 // both decode as nil, and both mean an empty-present list. The catalogue
323 // requires a present tabs slice, so a nil field becomes the empty list
324 // rather than a conversion failure.
325 if len(wireTabs) > catalogue.RemoteCatalogMaxTabsPerSess {
326 return catalogue.RemoteCatalogSession{}, ErrTooLarge
327 }
328 session.Tabs = make([]catalogue.RemoteCatalogTab, 0, len(wireTabs))
329 for i, wireTab := range wireTabs {
330 tab, err := catalogTabFromWire(wireTab)
331 if err != nil {
332 return catalogue.RemoteCatalogSession{}, err
333 }
334 if uint32(tab.Index) != uint32(i) {
335 return catalogue.RemoteCatalogSession{}, ErrInvalidMessage
336 }
337 session.Tabs = append(session.Tabs, tab)
338 }
339 if err := validateCatalogSession(session); err != nil {
340 return catalogue.RemoteCatalogSession{}, err
341 }
342 return session, nil
343}
344
345// validateCatalogSession enforces the exact catalogue contract of one session
346// so both encode and decode refuse a session that could not be published.
347func validateCatalogSession(session catalogue.RemoteCatalogSession) error {
348 catalog := catalogue.RemoteCatalog{
349 ProtocolVersion: protocol.Version,
350 SchemaVersion: catalogue.RemoteCatalogSchemaVersion,
351 Sessions: []catalogue.RemoteCatalogSession{session},
352 }
353 if err := catalogue.ValidateRemoteCatalog(catalog); err != nil {
354 return ErrInvalidMessage
355 }
356 return nil
357}
358
359// validateSnapshotDaemonFields enforces the per-daemon snapshot contract:
360// endpoint grammar, presentation origin, rank fit, closed availability and
361// failure taxonomies, and a syntactically present policy. The full port
362// projection rules (identity/incarnation pairing, observed version,
363// inventory, local authority) are re-checked by BrokerDaemonObservation
364// validation and, for remotes, the durable projection rule.
365func validateSnapshotDaemonFields(daemon ports.BrokerDaemonObservation) error {
366 if daemon.Local {
367 if daemon.Endpoint != "" {
368 return ErrInvalidMessage
369 }
370 } else if err := validateBrokerEndpoint(daemon.Endpoint); err != nil {
371 return err
372 }
373 if err := validateBrokerOrigin(daemon.DisplayOrigin); err != nil {
374 return err
375 }
376 if daemon.Rank < 0 || daemon.Rank > math.MaxUint32 {
377 return errConvertRange
378 }
379 if daemon.Availability < domain.RemoteAvailabilityUnknown ||
380 daemon.Availability > domain.RemoteAvailabilityNoDaemon {
381 return ErrInvalidMessage
382 }
383 if daemon.LastFailure.Kind > domain.RemoteFailureInvalidResponse {
384 return ErrInvalidMessage
385 }
386 return nil
387}
388
389// checkEnvEntries keeps the local error taxonomy at the adapter boundary.
390func checkEnvEntries(env []string) error {
391 return mapOptionalBrokerError(protoconv.BrokerEnvEntries(env))
392}